{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "chapter-intro",
   "metadata": {},
   "source": [
    "# 第 7 章：AI 智能体系统的架构与范式\n",
    "\n",
    "> 本笔记文件需要与《LangGraph实战》的第 7 章的内容配套使用。\n",
    "\n",
    "在本章中，我们将深入探索构建复杂、高效且可扩展的 AI 智能体系统的核心要素：架构设计与模式应用。随着我们不断提升 AI 智能体的能力，使其能够处理日益复杂的任务，仅仅依赖单一、线性的智能体模型已显得力不从心。如同城市规划需要蓝图、软件开发需要架构设计，构建强大的 AI 智能体系统同样需要精心设计的架构作为支撑。\n",
    "\n",
    "我们将从基础但至关重要的智能体工作流模式开始，例如提示链、路由、并行化、协调器-工作者和评估器-优化器，这些模式构成了构建复杂智能体行为的基石。随后，我们将逐步深入多智能体架构的世界，重点介绍主管架构和分层架构，揭示如何通过组织和协调多个专业智能体来提升系统的整体性能和可管理性。最后，我们将前瞻性地审视情境感知智能体架构，这种架构代表了 AI 智能体发展的新方向。\n",
    "\n",
    "通过本章的学习，您将不仅了解各种智能体系统的架构蓝图，更将掌握在 LangGraph 中实践这些架构模式的关键技术和方法。"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "environment-setup",
   "metadata": {},
   "source": [
    "### 🚀 环境准备\n",
    "\n",
    "首先加载必要的环境变量配置："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "cell-0",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "True"
      ]
     },
     "execution_count": 1,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from dotenv import load_dotenv\n",
    "\n",
    "load_dotenv()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "section-7-1",
   "metadata": {},
   "source": [
    "## 7.1 常见的智能体系统工作流\n",
    "\n",
    "在人工智能体开发领域，尤其是在 AI 智能体生态系统中，理解不同类型的智能体系统之间的细微差别至关重要。正如 Anthropic 在其对智能体构建模式的富有洞察力的分析中所强调的那样，我们可以更好地驾驭 AI 系统的复杂性。\n",
    "\n",
    "Anthropic 的研究突出了工作流（Workflow）和智能体（Agent）之间的一个关键区别：\n",
    "\n",
    "- **工作流**：LLM 和相关工具通过显式预定义的代码路径进行编排的系统\n",
    "- **智能体**：大语言模型（LLM）动态指导自身流程的系统，实时决策工具的使用以及实现目标所需的步骤\n",
    "\n",
    "在 LangGraph 的背景下，工作流使用其状态图架构优雅地实现，允许开发人员以可视化和编程方式定义系统中不同组件之间的信息和控制流。"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "section-7-1-1",
   "metadata": {},
   "source": [
    "### 7.1.1 工作流的基础构建模块：增强型 LLM\n",
    "\n",
    "现代 LLM 不仅仅是独立的模型，它们通过一系列功能得到增强，这些功能使它们能够与世界互动并执行超出简单文本生成的复杂任务。核心增强功能通常包括：\n",
    "\n",
    "- **检索（Retrieval）**：允许 LLM 访问和整合来自外部来源的信息\n",
    "- **工具（Tools）**：使 LLM 能够与外部系统交互并在现实世界中执行操作\n",
    "- **记忆（Memory）**：允许 LLM 保留和利用来自过去交互或工作流步骤的信息\n",
    "\n",
    "这些增强功能与核心 LLM 协同工作，构成了构建复杂工作流和智能体的基础。"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "section-7-1-2",
   "metadata": {},
   "source": [
    "### 7.1.2 提示链（Prompt Chaining）\n",
    "\n",
    "提示链是一种基本的工作流模式，专注于将复杂任务分解为一系列更简单、相互关联的步骤。在这种工作流中，一个 LLM 调用的输出成为后续调用的输入，从而创建一系列处理阶段。\n",
    "\n",
    "提示链的主要优势在于它能够通过简化每个 LLM 调用来提高准确性。通过将复杂任务分解为更小、更易于管理的子任务，每个 LLM 调用都会获得更集中且更明确的提示。"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "example-7-1",
   "metadata": {},
   "source": [
    "##### 示例 7-1：基于 Graph API 的提示链工作流实现\n",
    "\n",
    "首先导入必要的 LangGraph 和 LangChain 组件，并实现一个笑话生成和改进的提示链："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "cell-1",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "初始笑话：\n",
      "\n",
      "\n",
      "为什么猫不喜欢上网？  \n",
      "因为它们怕被“抓”住——  \n",
      "尤其是被键盘上的“猫爪”追着跑！🐾💻  \n",
      "\n",
      "（注：利用“抓”与“爪”的谐音，结合猫的天性，制造双关笑点）\n"
     ]
    }
   ],
   "source": [
    "from typing_extensions import TypedDict\n",
    "\n",
    "from langchain_openai import ChatOpenAI\n",
    "from langgraph.graph import StateGraph, START, END\n",
    "\n",
    "\n",
    "# 使用 TypedDict 定义图状态，用于类型提示和状态管理\n",
    "class State(TypedDict):\n",
    "    topic: str\n",
    "    joke: str\n",
    "    improved_joke: str\n",
    "    final_joke: str\n",
    "\n",
    "# 配置模型\n",
    "llm = ChatOpenAI(model=\"Qwen/Qwen3-8B\")\n",
    "\n",
    "# 图中的节点，每个节点代表提示链中的一个步骤\n",
    "def generate_joke(state: State):\n",
    "    \"\"\"第一个 LLM 调用，根据主题生成初始笑话\"\"\"\n",
    "    msg = llm.invoke(f\"写一个关于 {state['topic']} 的简短笑话\") # 使用状态中的主题调用 LLM\n",
    "    return {\"joke\": msg.content} # 返回生成的笑话，更新状态中的 'joke' 键\n",
    "\n",
    "def check_punchline(state: State):\n",
    "    \"\"\"门控函数，检查笑话是否有妙语\"\"\"\n",
    "    # 简单检查 - 笑话是否包含 \"?\" 或 \"!\" 作为妙语存在的代理\n",
    "    if \"?\" in state[\"joke\"] or \"!\" in state[\"joke\"] or \"？\" in state[\"joke\"] or \"！\" in state[\"joke\"]:\n",
    "        return \"Pass\" # 笑话通过妙语检查\n",
    "    return \"Fail\" # 笑话未能通过妙语检查\n",
    "\n",
    "def improve_joke(state: State):\n",
    "    \"\"\"第二个 LLM 调用，通过添加文字游戏来改进笑话\"\"\"\n",
    "    msg = llm.invoke(f\"通过添加文字游戏使这个笑话更有趣：{state['joke']}\") # 调用 LLM 来改进笑话\n",
    "    return {\"improved_joke\": msg.content} # 返回改进后的笑话，更新状态中的 'improved_joke'\n",
    "\n",
    "def polish_joke(state: State):\n",
    "    \"\"\"第三个 LLM 调用，用于最终润色，添加令人惊讶的转折\"\"\"\n",
    "    msg = llm.invoke(f\"为这个笑话添加一个令人惊讶的转折：{state['improved_joke']}\") # 调用 LLM 来润色笑话\n",
    "    return {\"final_joke\": msg.content} # 返回润色后的笑话，更新状态中的 'final_joke'\n",
    "\n",
    "# 使用 StateGraph 构建工作流，使用定义的状态进行初始化\n",
    "workflow = StateGraph(State)\n",
    "\n",
    "# 将节点添加到工作流图中，将它们与定义的函数关联起来\n",
    "workflow.add_node(\"generate_joke\", generate_joke)\n",
    "workflow.add_node(\"improve_joke\", improve_joke)\n",
    "workflow.add_node(\"polish_joke\", polish_joke)\n",
    "\n",
    "# 定义边缘以连接节点并建立工作流序列\n",
    "workflow.add_edge(START, \"generate_joke\") # 开始节点连接到 'generate_joke' 节点\n",
    "workflow.add_conditional_edges(\n",
    "    \"generate_joke\", check_punchline, {\"Fail\": \"improve_joke\", \"Pass\": END} # 'generate_joke' 之后的条件边缘，基于 'check_punchline' 输出\n",
    ")\n",
    "workflow.add_edge(\"improve_joke\", \"polish_joke\") # 'improve_joke' 节点连接到 'polish_joke' 节点\n",
    "workflow.add_edge(\"polish_joke\", END) # 'polish_joke' 节点连接到结束节点\n",
    "\n",
    "# 将工作流图编译为可执行链\n",
    "chain = workflow.compile()\n",
    "\n",
    "\n",
    "# 使用初始状态（主题：\"cats\"）调用编译链\n",
    "state = chain.invoke({\"topic\": \"cats\"})\n",
    "print(\"初始笑话：\")\n",
    "print(state[\"joke\"])\n",
    "\n",
    "if \"improved_joke\" in state: # 检查 'improved_joke' 是否存在于状态中，指示妙语检查失败\n",
    "    print(\"\\n改进后的笑话：\")\n",
    "    print(state[\"improved_joke\"])\n",
    "    print(\"\\n最终笑话：\")\n",
    "    print(state[\"final_joke\"])"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "explanation-7-1",
   "metadata": {},
   "source": [
    "**💡 核心概念解析**：\n",
    "\n",
    "在这段代码中，我们展示了提示链的关键特性：\n",
    "\n",
    "- **顺序处理**：每个步骤的输出成为下一个步骤的输入\n",
    "- **门控机制**：`check_punchline` 函数作为质量控制检查点\n",
    "- **条件分支**：根据门控结果决定是直接结束还是继续改进\n",
    "- **状态管理**：使用 TypedDict 清晰定义数据流"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "example-7-2",
   "metadata": {},
   "source": [
    "##### 示例 7-2：基于 Functional API 的提示链工作流实现\n",
    "\n",
    "接下来让我们看看如何使用 LangGraph 的 Functional API 实现相同的提示链逻辑："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "cell-2",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "**笑话：**  \n",
      "为什么猫不喜欢当数学老师？  \n",
      "因为它们总把“1”和“0”弄混——一个“喵”（1），一个“溜”（0）！ 😺  \n",
      "\n",
      "**解析：**  \n",
      "利用猫的叫声“喵”和数字“1”的谐音，以及“溜”与“0”的字形关联，制造双关笑点，既幽默又贴近猫咪的日常行为，简短易懂！\n"
     ]
    }
   ],
   "source": [
    "from langchain_openai import ChatOpenAI\n",
    "from langgraph.func import entrypoint, task\n",
    "\n",
    "llm = ChatOpenAI(model=\"Qwen/Qwen3-8B\")\n",
    "\n",
    "# 使用 @task 装饰器定义的任务，代表工作流中的步骤\n",
    "@task\n",
    "def generate_joke(topic: str):\n",
    "    \"\"\"第一个 LLM 调用，生成初始笑话\"\"\"\n",
    "    msg = llm.invoke(f\"写一个关于 {topic} 的简短笑话\") # 调用 LLM 以根据主题生成笑话\n",
    "    return msg.content # 返回生成的笑话\n",
    "\n",
    "def check_punchline(joke: str):\n",
    "    \"\"\"门控函数，检查笑话是否有妙语\"\"\"\n",
    "    # 简单检查 - 笑话是否包含 \"?\" 或 \"!\"\n",
    "    if \"?\" in joke or \"!\" in joke:\n",
    "        return \"Fail\" # 笑话未能通过妙语检查\n",
    "    return \"Pass\" # 笑话通过妙语检查\n",
    "\n",
    "@task\n",
    "def improve_joke(joke: str):\n",
    "    \"\"\"第二个 LLM 调用，改进笑话\"\"\"\n",
    "    msg = llm.invoke(f\"通过添加文字游戏使这个笑话更有趣：{joke}\") # 调用 LLM 以改进笑话\n",
    "    return msg.content # 返回改进后的笑话\n",
    "\n",
    "@task\n",
    "def polish_joke(joke: str):\n",
    "    \"\"\"第三个 LLM 调用，用于最终润色\"\"\"\n",
    "    msg = llm.invoke(f\"为这个笑话添加一个令人惊讶的转折：{joke}\") # 调用 LLM 以润色笑话\n",
    "    return msg.content # 返回润色后的笑话\n",
    "\n",
    "# 入口点装饰函数使用 Functional API 定义工作流\n",
    "@entrypoint()\n",
    "def workflow(topic: str):\n",
    "    original_joke = generate_joke(topic).result() # 执行 'generate_joke' 任务\n",
    "    if check_punchline(original_joke) == \"Pass\": # 基于 'check_punchline' 输出的条件检查\n",
    "        return original_joke # 如果妙语检查通过，则返回原始笑话\n",
    "\n",
    "    improved_joke = improve_joke(original_joke).result() # 如果妙语检查失败，则执行 'improve_joke' 任务\n",
    "    return polish_joke(improved_joke).result() # 执行 'polish_joke' 任务并返回最终结果\n",
    "\n",
    "# 调用工作流\n",
    "state = workflow.invoke(\"cats\")\n",
    "print(state)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "explanation-7-2",
   "metadata": {},
   "source": [
    "**💡 Functional API 的优势**：\n",
    "\n",
    "- **更简洁的代码**：使用 `@task` 装饰器和函数式编程风格\n",
    "- **类型安全**：函数参数提供清晰的类型提示\n",
    "- **易于测试**：每个任务都是独立的函数，便于单元测试\n",
    "- **灵活的控制流**：使用 Python 的原生控制结构"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "section-7-1-3",
   "metadata": {},
   "source": [
    "### 7.1.3 路由 (Routing)\n",
    "\n",
    "路由工作流旨在通过对输入进行分类并将其定向到专门的下游任务来处理各种输入。当处理需要处理各种输入类型的复杂应用程序时，此模式尤其有价值，每种输入类型都需要不同的处理方法。\n",
    "\n",
    "路由背后的核心思想是实施决策步骤，该步骤分析输入并确定最合适的后续处理路径。这允许关注点分离，从而可以为每个输入类别开发更集中和优化的提示和流程。"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "example-7-3",
   "metadata": {},
   "source": [
    "##### 示例 7-3：基于 Graph API 的路由工作流实现\n",
    "\n",
    "让我们实现一个智能内容路由器，能够根据用户输入生成故事、笑话或诗歌："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "cell-3",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "“有一天，我带猫去理发店。店员问：‘剪哪部分？’猫一言不发，只是‘喵’了一声。店员愣住：‘你这是要剪掉毛，还是剪掉猫？’猫认真地说：‘不，我要剪掉‘毛’线——别误会，我不是在说光纤，是说‘猫’线！’” \n",
      "\n",
      "（笑点在于猫用“毛线”自称“猫线”，既指它自己的毛，又暗示它习惯性地“动”线，谐音双关，同时带点猫的调皮性格。）\n"
     ]
    }
   ],
   "source": [
    "import json\n",
    "\n",
    "from langchain_core.messages import HumanMessage, SystemMessage\n",
    "from langchain_openai import ChatOpenAI\n",
    "from langgraph.graph import StateGraph, START, END\n",
    "\n",
    "llm = ChatOpenAI(model=\"Qwen/Qwen3-8B\")\n",
    "\n",
    "# 路由工作流的状态定义\n",
    "class State(TypedDict):\n",
    "    input: str\n",
    "    decision: str\n",
    "    output: str\n",
    "\n",
    "# 图中的节点，每个节点处理特定的路由 (story, joke, poem)\n",
    "def llm_call_1(state: State):\n",
    "    \"\"\"写一个故事\"\"\"\n",
    "    result = llm.invoke(state[\"input\"]) # 调用 LLM 根据输入写一个故事\n",
    "    return {\"output\": result.content} # 返回故事，更新状态中的 'output'\n",
    "\n",
    "def llm_call_2(state: State):\n",
    "    \"\"\"写一个笑话\"\"\"\n",
    "    result = llm.invoke(state[\"input\"]) # 调用 LLM 根据输入写一个笑话\n",
    "    return {\"output\": result.content} # 返回笑话，更新状态中的 'output'\n",
    "\n",
    "def llm_call_3(state: State):\n",
    "    \"\"\"写一首诗\"\"\"\n",
    "    result = llm.invoke(state[\"input\"]) # 调用 LLM 根据输入写一首诗\n",
    "    return {\"output\": result.content} # 返回诗歌，更新状态中的 'output'\n",
    "\n",
    "def llm_call_router(state: State):\n",
    "    \"\"\"使用结构化输出将输入路由到适当的节点\"\"\"\n",
    "    # 使用结构化输出调用增强型 LLM，以充当路由逻辑\n",
    "    model = ChatOpenAI(model=\"Qwen/Qwen2.5-7B-Instruct\", model_kwargs={ \"response_format\": { \"type\": \"json_object\" } })\n",
    "    ai_msg = model.invoke(\n",
    "        [\n",
    "            SystemMessage(\n",
    "                content=\"You are a router that directs user input to the appropriate handler. Return a JSON object with a 'step' key and one of these values: 'story', 'joke', or 'poem'. For example: {'step': 'joke'}\" # 路由 LLM 的系统消息\n",
    "            ),\n",
    "            HumanMessage(content=state[\"input\"]), # 用户输入消息\n",
    "        ]\n",
    "    )\n",
    "    decision = json.loads(ai_msg.content)\n",
    "    return {\"decision\": decision[\"step\"]} # 返回路由决策，更新状态中的 'decision'\n",
    "\n",
    "# 条件边缘函数，根据决策路由到适当的节点\n",
    "def route_decision(state: State):\n",
    "    # 根据状态中的 'decision' 返回您想要访问的下一个节点名称\n",
    "    if state[\"decision\"] == \"story\":\n",
    "        return \"llm_call_1\"\n",
    "    elif state[\"decision\"] == \"joke\":\n",
    "        return \"llm_call_2\"\n",
    "    elif state[\"decision\"] == \"poem\":\n",
    "        return \"llm_call_3\"\n",
    "\n",
    "# 使用 StateGraph 构建路由工作流\n",
    "router_builder = StateGraph(State)\n",
    "\n",
    "# 将节点添加到图中\n",
    "router_builder.add_node(\"llm_call_1\", llm_call_1)\n",
    "router_builder.add_node(\"llm_call_2\", llm_call_2)\n",
    "router_builder.add_node(\"llm_call_3\", llm_call_3)\n",
    "router_builder.add_node(\"llm_call_router\", llm_call_router)\n",
    "\n",
    "# 定义边缘以连接节点并建立路由逻辑\n",
    "router_builder.add_edge(START, \"llm_call_router\") # 开始节点连接到路由器节点\n",
    "router_builder.add_conditional_edges(\n",
    "    \"llm_call_router\",\n",
    "    route_decision,\n",
    "    {  # 由 route_decision 返回的名称：要访问的下一个节点的名称\n",
    "        \"llm_call_1\": \"llm_call_1\",\n",
    "        \"llm_call_2\": \"llm_call_2\",\n",
    "        \"llm_call_3\": \"llm_call_3\",\n",
    "    },\n",
    ") # 从路由器到专用节点的条件边缘，基于路由决策\n",
    "router_builder.add_edge(\"llm_call_1\", END) # 专用节点连接到结束节点\n",
    "router_builder.add_edge(\"llm_call_2\", END)\n",
    "router_builder.add_edge(\"llm_call_3\", END)\n",
    "\n",
    "# 编译路由工作流图\n",
    "router_workflow = router_builder.compile()\n",
    "\n",
    "# 使用示例输入调用路由工作流\n",
    "state = router_workflow.invoke({\"input\": \"给我写一个关于猫的笑话\"})\n",
    "print(state[\"output\"]) "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "explanation-7-3",
   "metadata": {},
   "source": [
    "**💡 路由机制关键特性**：\n",
    "\n",
    "- **智能分类**：路由器节点使用 LLM 的理解能力来分析输入意图\n",
    "- **结构化输出**：使用 JSON 格式确保路由决策的可靠性\n",
    "- **专门处理**：每个路由目标都有专门的处理逻辑\n",
    "- **可扩展性**：易于添加新的路由目标和处理器"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "example-7-4",
   "metadata": {},
   "source": [
    "##### 示例 7-4：基于 Functional API 的路由工作流实现\n",
    "\n",
    "让我们用 Functional API 实现相同的路由逻辑："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "cell-4",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{'router_workflow': <Future at 0x1308497d0 state=pending>}\n",
      "\n",
      "\n",
      "{'llm_call_2': '当然可以！这里有一个关于猫的笑话：\\n\\n为什么猫不喜欢打网球？\\n\\n因为它们担心会捡不到球！'}\n",
      "\n",
      "\n"
     ]
    }
   ],
   "source": [
    "import json\n",
    "\n",
    "from langchain_openai import ChatOpenAI\n",
    "from langchain_core.messages import HumanMessage, SystemMessage\n",
    "from langgraph.func import entrypoint, task\n",
    "\n",
    "llm = ChatOpenAI(model=\"Qwen/Qwen2.5-7B-Instruct\")\n",
    "\n",
    "@task\n",
    "def llm_call_1(input: str):\n",
    "    \"\"\"写一个故事\"\"\"\n",
    "    result = llm.invoke(input) # 调用 LLM 根据输入写一个故事\n",
    "    return result.content # 返回故事\n",
    "\n",
    "@task\n",
    "def llm_call_2(input: str):\n",
    "    \"\"\"写一个笑话\"\"\"\n",
    "    result = llm.invoke(input) # 调用 LLM 根据输入写一个笑话\n",
    "    return result.content # 返回笑话\n",
    "\n",
    "@task\n",
    "def llm_call_3(input: str):\n",
    "    \"\"\"写一首诗\"\"\"\n",
    "    result = llm.invoke(input) # 调用 LLM 根据输入写一首诗\n",
    "    return result.content # 返回诗歌\n",
    "\n",
    "def llm_call_router(input: str):\n",
    "    \"\"\"使用结构化输出将输入路由到适当的节点\"\"\"\n",
    "    # 使用结构化输出调用增强型 LLM，以充当路由逻辑\n",
    "    model = ChatOpenAI(model=\"Qwen/Qwen2.5-7B-Instruct\", model_kwargs={ \"response_format\": { \"type\": \"json_object\" } })\n",
    "    ai_msg = model.invoke(\n",
    "        [\n",
    "            SystemMessage(\n",
    "                content=\"You are a router that directs user input to the appropriate handler. Return a JSON object with a 'step' key and one of these values: 'story', 'joke', or 'poem'. For example: {'step': 'joke'}\" # 路由 LLM 的系统消息\n",
    "            ),\n",
    "            HumanMessage(content=input), # 用户输入消息\n",
    "        ]\n",
    "    )\n",
    "    decision = json.loads(ai_msg.content)\n",
    "    return {\"decision\": decision[\"step\"]} # 返回路由决策\n",
    "\n",
    "# 入口点装饰函数定义路由工作流\n",
    "@entrypoint()\n",
    "def router_workflow(input: str):\n",
    "    next_step = llm_call_router(input)[\"decision\"] # 获取路由决策的 'decision' 值\n",
    "    llm_call = None # 初始化 llm_call 变量\n",
    "    \n",
    "    if next_step == \"story\": # 基于路由器决策的条件路由\n",
    "        llm_call = llm_call_1 # 如果路由是 'story'，则分配 'llm_call_1' 任务\n",
    "    elif next_step == \"joke\":\n",
    "        llm_call = llm_call_2 # 如果路由是 'joke'，则分配 'llm_call_2' 任务\n",
    "    elif next_step == \"poem\":\n",
    "        llm_call = llm_call_3 # 如果路由是 'poem'，则分配 'llm_call_3' 任务\n",
    "        \n",
    "    if llm_call is None:\n",
    "        raise ValueError(f\"Invalid routing decision: {next_step}\")\n",
    "        \n",
    "    return llm_call(input) # 执行选定的 LLM 调用任务并返回结果\n",
    "\n",
    "# 调用路由工作流\n",
    "for step in router_workflow.stream(\"给我写一个关于猫的笑话\", stream_mode=\"updates\"):\n",
    "    print(step)\n",
    "    print(\"\\n\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "section-7-1-4",
   "metadata": {},
   "source": [
    "### 7.1.4 并行化 (Parallelization)\n",
    "\n",
    "并行化是另一种工作流模式，它利用增强型 LLM 同时处理任务不同方面的能力。并行化不是按顺序处理任务，而是允许同时进行 LLM 调用，其输出稍后以编程方式聚合。\n",
    "\n",
    "这种方法可以体现在两个主要变体中：\n",
    "- **分段 (Sectioning)**：将任务分解为可以并行执行的独立子任务\n",
    "- **投票 (Voting)**：多次运行相同的任务以获得更稳健和可靠的结果\n",
    "\n",
    "并行化的主要好处是提高了效率，尤其是在子任务真正独立并且可以并发处理的情况下。"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "example-7-5",
   "metadata": {},
   "source": [
    "##### 示例 7-5：基于 Graph API 的并行化工作流实现\n",
    "\n",
    "让我们实现一个并行内容生成工作流，同时生成故事、笑话和诗歌："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "cell-5",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "这是一个关于 cats 的故事、笑话和诗歌！\n",
      "\n",
      "故事：\n",
      "\n",
      "\n",
      "好的，我将按照您的要求创作一个关于猫咪的故事。为了让故事更生动，我会先整理一些基本框架。请您看看以下内容是否符合您的预期。如果您有其他想法，可以随时提出，我会进行调整。\n",
      "\n",
      "### 故事梗概\n",
      "\n",
      "- **孤独的流浪猫**：在废弃工厂的阴影下，橘猫\"琥珀\"独自生活，以捕鼠为生却始终渴望温暖的家。\n",
      "- **神秘的相遇**：意外发现废弃仓库中藏着会发光的蓝猫，对方对琥珀的特殊关注让它们产生羁绊。\n",
      "- **意外的转折**：蓝猫透露自己是被遗弃的实验品，琥珀在帮助它逃离的过程中发现工厂的秘密。\n",
      "- **真相的重逢**：多年后成为宠物的琥珀，与当年的蓝猫在同一个花园重逢，见证彼此的蜕变。\n",
      "\n",
      "---\n",
      "锈蚀的铁门在夜风中发出呜咽，琥珀蹲坐在门框上，尾巴有节奏地拍打着生锈的门栓。这是她在这座废弃工厂生活的第七个冬天，月光从破碎的玻璃窗斜切进来，在满地尘埃上织出银丝般的光网。\n",
      "\n",
      "她用爪子拨开蛛网，露出墙角堆积的纸箱。那些曾经装满油桶的箱子现在成了她的床铺，散发着机油和旧报纸混合的气味。琥珀的橘色毛发沾满灰尘，却在月光下泛着琥珀色的光晕，像一块凝固的阳光。\n",
      "\n",
      "\"喵——\"\n",
      "\n",
      "远处传来细微的呜咽，琥珀竖起耳朵。声音来自工厂西侧的通风管道，那里的铁皮被老鼠啃出孔洞，她记得上周还看到一只花斑猫被困在那儿。这次的声音不一样，像是金属摩擦的铮鸣，又带着某种机械的韵律。\n",
      "\n",
      "她轻巧地跃上锈蚀的钢架，爪子陷入冰冷的铁锈里。在管道深处，一只蓝猫正用尾巴卷着半截生锈的铁丝，琥珀注意到它左耳缺了一块，像被什么利器划过。蓝猫突然转头，琥珀这才发现它的眼睛是琥珀色的，和自己一样。\n",
      "\n",
      "\"你也是逃出来的？\"琥珀试探着靠近，冰凉的铁丝在爪间发出细碎的声响。\n",
      "\n",
      "蓝猫的尾巴轻轻摆动，发出类似风铃的清脆声：\"我是被关在这里的实验品。\"它用前爪抚过腹部的金属鳞片，\"他们用特殊合金覆盖我的皮肤，说这样能让我在黑暗中发光。\"\n",
      "\n",
      "琥珀的瞳孔突然收缩。她记得三个月前，工厂东侧的仓库里出现过异常的蓝光，那只被遗弃的实验猫在月光下闪烁着幽蓝的光，像坠落的星星。当时她本能地扑向那只猫，却被铁丝网拦住，现在才明白那不是普通的流浪猫。\n",
      "\n",
      "\"他们每天用高温喷射器清洗我的鳞片，\"蓝猫的声音带着金属的震颤，\"疼痛就像融化的铁水在皮肤里流淌。\"它突然用尾巴卷住琥珀的颈项，\"但你不一样，你的毛发会吸收月光。\"\n",
      "\n",
      "琥珀感觉到自己毛发中的温暖在流动，那是她每天在月光下梳理时体会到的奇迹。她想起自己被遗弃在垃圾站的那天，正是月光最盛的时候，漆黑的垃圾袋里，她发现自己的毛发在暗处发亮。\n",
      "\n",
      "\"要带我出去吗？\"蓝猫用爪子碰了碰琥珀的下巴，\"我知道通风管道尽头有个废弃的配电室。\"\n",
      "\n",
      "琥珀的尾巴突然剧烈抖动，她想起上周在配电室看到的场景：无数蓝色光点在黑暗中闪烁，像被囚禁的萤火虫。那些光点正在疯狂地跳动，她的瞳孔倒映出诡异的图案。\n",
      "\n",
      "当蓝猫在凌晨三点带她钻进配电室时，琥珀看到墙上密密麻麻的注射痕迹。那些蓝色光点原来是被困在玻璃罐中的实验体，每个罐子都贴着\"蓝光猫\"的标签。她突然明白自己身上的光亮并非偶然，而是某个未完成实验的余韵。\n",
      "\n",
      "\"他们把我的同类都装在罐子里，\"蓝猫的声音开始发抖，\"我偷了钥匙，但还需要你帮我引开保安。\"它腹部的金属鳞片反射着月色，\"我的光亮只能维持到黎明。\"\n",
      "\n",
      "琥珀在月光下跳起，用爪子掀翻一个装满老鼠的铁皮桶。老鼠们惊慌逃窜的声响在空旷的工厂里回荡，蓝猫趁机钻出通风管道。当保安的脚步声渐渐远去，琥珀发现蓝猫的鳞片正在褪色，它的身体变得透明。\n",
      "\n",
      "\"快走！\"蓝猫突然将一团蓝色光点撒向琥珀，\"这些是我最后的光。\"在光点的指引下，琥珀带着蓝猫穿过废弃的生产线，工业遗迹的阴影里，她第一次感受到被拥抱的温度。\n",
      "\n",
      "三年后，琥珀坐在落地窗前的飘窗上，看着楼下花园里玩耍的孩子。夕阳为她的橘色毛发镀上金边，而窗外的月光下，一只蓝猫正悠闲地舔着爪子。它腹部的鳞片在暮色中泛着微光，像被揉碎的星尘。\n",
      "\n",
      "\"还记得那座工厂吗？\"蓝猫突然开口，声音不再机械。琥珀的尾巴轻轻扫过窗台，她记得那个夜晚，月光为它们编织的银网里，藏着关于自由的密码。\n",
      "\n",
      "---\n",
      "\n",
      "**接下来故事可能会沿着这些方向发展**：\n",
      "\n",
      "- **金属记忆**：蓝猫腹部的鳞片暗示其可能具有特殊的记忆存储能力，为后续揭示实验室秘密埋下伏笔\n",
      "\n",
      "- **光之传承**：琥珀发现的光亮特性，暗示猫群可能存在某种未知的基因传承现象\n",
      "\n",
      "- **时间回响**：工厂墙面的注射痕迹与蓝猫褪色的鳞片，暗示实验对时间感知的影响\n",
      "\n",
      "希望这个故事能满足您对猫咪的想象。如果需要更多细节或调整故事走向，请随时告诉我。\n",
      "\n",
      "笑话：\n",
      "\n",
      "\n",
      "当然可以！这里有一个关于猫的笑话：\n",
      "\n",
      "---\n",
      "\n",
      "**猫的哲学**  \n",
      "一只猫坐在沙发上，看着窗外的鸟儿自由飞翔，叹了口气说：“如果我有翅膀，大概率会飞走吧。”  \n",
      "它的主人好奇地问：“那你为什么还不飞？”  \n",
      "猫眨眨眼，认真回复：“因为我的爪子已经够厉害了，飞走太没挑战性了。”  \n",
      "主人又问：“那你想做什么？”  \n",
      "猫耸耸肩：“找个地方，让人类觉得我神秘又高贵——比如，假装在思考人生。”  \n",
      "主人：“……你真的在思考人生吗？”  \n",
      "猫：“当然，我在思考如何让沙发更舒服，同时还能让人类以为我是在‘进化’。”  \n",
      "\n",
      "---\n",
      "\n",
      "**笑点解析**  \n",
      "猫的“哲学”其实是一种拟人化幽默，用人类的思维模式调侃猫的日常行为（比如抓沙发、装酷），同时结合“神秘感”和“进化”等词制造反差，让听众在猫的“高智商”调侃中发笑。  \n",
      "（如果想更简洁，也可以改成：  \n",
      "“猫：我如果会飞，大概率会飞走。  \n",
      "主人：那你为什么不飞？  \n",
      "猫：因为我的爪子已经够厉害了，飞走太没意思了。”）  \n",
      "\n",
      "需要更多风格（比如冷笑话、谐音梗、情景剧）可以告诉我！ 😸\n",
      "\n",
      "诗歌：\n",
      "\n",
      "\n",
      "《猫的独白》\n",
      "\n",
      "我是一团被月光揉皱的云\n",
      "在窗台踱步时\n",
      "总把影子拉成\n",
      "从不闭合的门\n",
      "\n",
      "它们用瞳孔丈量黑暗\n",
      "每一次眨眼都是\n",
      "对光的叛逃\n",
      "也是对永恒的臣服\n",
      "\n",
      "我曾是钟表里\n",
      "被遗忘的齿轮\n",
      "直到学会在寂静中\n",
      "用爪尖写下诗行\n",
      "\n",
      "当人类在梦里沉溺\n",
      "我们便成为\n",
      "黑夜的褶皱\n",
      "藏着所有未说出口的词\n",
      "\n",
      "（注：通过意象叠加和隐喻手法，将猫的神秘特质与人类情感交织。月光、影子、瞳孔等元素构建出超现实的画面感，同时\"被遗忘的齿轮\"暗示猫与机械世界的疏离，\"未说出口的词\"则暗喻它们所承载的深层情感与语言之外的智慧。）\n"
     ]
    }
   ],
   "source": [
    "from typing_extensions import TypedDict\n",
    "from langchain_openai import ChatOpenAI\n",
    "from langgraph.graph import StateGraph, START, END\n",
    "\n",
    "# 并行工作流的图状态定义\n",
    "class State(TypedDict):\n",
    "    topic: str\n",
    "    joke: str\n",
    "    story: str\n",
    "    poem: str\n",
    "    combined_output: str\n",
    "\n",
    "llm = ChatOpenAI(model=\"Qwen/Qwen3-8B\")\n",
    "\n",
    "# 图中的节点，每个节点并行生成不同类型的内容\n",
    "def call_llm_1(state: State):\n",
    "    \"\"\"第一个 LLM 调用，生成初始笑话\"\"\"\n",
    "    msg = llm.invoke(f\"写一个关于 {state['topic']} 的笑话\") # 调用 LLM 以根据主题写一个笑话\n",
    "    return {\"joke\": msg.content} # 返回笑话，更新状态中的 'joke'\n",
    "\n",
    "def call_llm_2(state: State):\n",
    "    \"\"\"第二个 LLM 调用，生成故事\"\"\"\n",
    "    msg = llm.invoke(f\"写一个关于 {state['topic']} 的故事\") # 调用 LLM 以根据主题写一个故事\n",
    "    return {\"story\": msg.content} # 返回故事，更新状态中的 'story'\n",
    "\n",
    "def call_llm_3(state: State):\n",
    "    \"\"\"第三个 LLM 调用，生成诗歌\"\"\"\n",
    "    msg = llm.invoke(f\"写一首关于 {state['topic']} 的诗歌\") # 调用 LLM 以根据主题写一首诗\n",
    "    return {\"poem\": msg.content} # 返回诗歌，更新状态中的 'poem'\n",
    "\n",
    "def aggregator(state: State):\n",
    "    \"\"\"将笑话、故事和诗歌组合成单个输出\"\"\"\n",
    "    combined = f\"这是一个关于 {state['topic']} 的故事、笑话和诗歌！\\n\\n\" # 开始组合输出\n",
    "    combined += f\"故事：\\n{state['story']}\\n\\n\" # 将故事添加到组合输出\n",
    "    combined += f\"笑话：\\n{state['joke']}\\n\\n\" # 将笑话添加到组合输出\n",
    "    combined += f\"诗歌：\\n{state['poem']}\" # 将诗歌添加到组合输出\n",
    "    return {\"combined_output\": combined} # 返回组合输出，更新状态中的 'combined_output'\n",
    "\n",
    "# 使用 StateGraph 构建并行工作流\n",
    "parallel_builder = StateGraph(State)\n",
    "\n",
    "# 将节点添加到图中\n",
    "parallel_builder.add_node(\"call_llm_1\", call_llm_1)\n",
    "parallel_builder.add_node(\"call_llm_2\", call_llm_2)\n",
    "parallel_builder.add_node(\"call_llm_3\", call_llm_3)\n",
    "parallel_builder.add_node(\"aggregator\", aggregator)\n",
    "\n",
    "# 定义边缘以连接节点并建立并行执行\n",
    "parallel_builder.add_edge(START, \"call_llm_1\") # 开始节点连接到 'call_llm_1' 以进行并行执行\n",
    "parallel_builder.add_edge(START, \"call_llm_2\") # 开始节点连接到 'call_llm_2' 以进行并行执行\n",
    "parallel_builder.add_edge(START, \"call_llm_3\") # 开始节点连接到 'call_llm_3' 以进行并行执行\n",
    "parallel_builder.add_edge(\"call_llm_1\", \"aggregator\") # 'call_llm_1' 节点在完成后连接到聚合器\n",
    "parallel_builder.add_edge(\"call_llm_2\", \"aggregator\") # 'call_llm_2' 节点在完成后连接到聚合器\n",
    "parallel_builder.add_edge(\"call_llm_3\", \"aggregator\") # 'call_llm_3' 节点在完成后连接到聚合器\n",
    "parallel_builder.add_edge(\"aggregator\", END) # 聚合器节点连接到结束节点\n",
    "\n",
    "# 编译并行工作流图\n",
    "parallel_workflow = parallel_builder.compile()\n",
    "\n",
    "# 使用示例输入调用并行工作流\n",
    "state = parallel_workflow.invoke({\"topic\": \"cats\"})\n",
    "print(state[\"combined_output\"]) "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "explanation-7-5",
   "metadata": {},
   "source": [
    "**💡 并行化的关键优势**：\n",
    "\n",
    "- **效率提升**：多个 LLM 调用同时执行，减少总体等待时间\n",
    "- **独立处理**：每个任务专注于特定方面，避免上下文混乱\n",
    "- **结果聚合**：通过聚合器节点组合所有结果\n",
    "- **可扩展性**：易于添加更多并行处理分支"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "example-7-6",
   "metadata": {},
   "source": [
    "##### 示例 7-6：基于 Functional API 的并行化工作流实现\n",
    "\n",
    "让我们用 Functional API 实现相同的并行处理逻辑："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "cell-6",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{'call_llm_3': '\\n\\n《猫的独白》\\n\\n午夜的剪影里\\n你的瞳孔收缩成两枚铜币\\n轻轻一跃\\n便撕开丝绸的帷幕\\n\\n我数着你胡须的针脚\\n编织成雾的经纬\\n当月光在瓦罐里发酵\\n你衔着影子踱步\\n\\n琥珀色的沉默在掌心\\n凝结成未拆封的信\\n你用尾巴丈量时间\\n在墙角画出螺旋的痕\\n\\n最古老的谜题\\n被你藏进毛茸茸的爪印\\n当人类还在笨拙地学习\\n如何用目光驯服星辰'}\n",
      "\n",
      "\n",
      "{'call_llm_1': '\\n\\n好的，这里有一个关于猫的笑话：\\n\\n---\\n\\n**笑话：**\\n\\n一只猫对主人说：“我想去太空旅行，成为宇航员！”\\n\\n主人惊讶地问：“你有这方面的资质吗？”\\n\\n猫骄傲地甩了甩尾巴：“当然！我有完美的身体结构，锋利的爪子能抓握太空舱，还有夜视能力——毕竟在宇宙里，白天太亮了，我需要隐身！”\\n\\n主人想了想，指着电视上的火箭发射直播：“那先从这开始吧，你跳上去试试？”\\n\\n猫盯着屏幕，不屑地说：“这太简单了……我只需要一个铲子和一盒猫粮，就能在月球上开一家‘喵星人零食店’。”\\n\\n结果，它跳上沙发，用爪子拍了拍遥控器，把电视节目换成了猫咪视频合集。\\n\\n---\\n\\n**笑点解析：**  \\n猫的“太空梦想”被主人用现实的方式打脸（遥控器），而猫的回应又展现了它的傲娇与自我中心，最后结局反转幽默，同时暗含对猫日常行为的调侃（笑脸、铲子、猫粮这些元素都是猫主子熟悉的“战场”）。'}\n",
      "\n",
      "\n",
      "{'call_llm_2': '\\n\\n好的，我将按照您的要求创作一个关于猫咪的故事。为了让故事更生动，我会先梳理一些基本的情节脉络。请您看看以下内容是否符合您的预期。如果您有其他想法，可以随时提出，我会进行调整。\\n\\n### 故事梗概\\n\\n- **神秘相遇**：在暴雨夜，一只玳瑁猫闯入老旧的图书馆，带着不属于这个时代的银色项圈，暗示着不寻常的身世。\\n\\n- **往事重现**：女主人林夕在整理旧书时，发现猫咪项圈上的刻痕与父亲留下的日记本记载惊人吻合，揭开尘封的往事。\\n\\n- **时空穿梭**：猫咪似乎能感知时间的流动，通过不同年代的书籍带林夕穿越，重现父亲与猫咪的往事与抉择。\\n\\n- **命运重叠**：当林夕发现父亲曾用同样的方式寻找失踪的猫咪，她终于理解生命与记忆的传承，选择与猫咪共同守护图书馆。\\n\\n- **永恒守望**：图书馆成为连接不同时空的纽带，猫咪在每个雨夜守护着这座建筑，见证着新旧交替与永恒的爱。\\n\\n---\\n暴雨砸在图书馆的彩绘玻璃上，发出清脆的敲击声。我握着《猫与神秘的十二重门》这本书，突然听见阁楼传来窸窣响动。推开吱呀作响的木门，一团毛茸茸的影子正蹲在古董书架前，琥珀色的眼睛在昏暗中泛着微光。\\n\\n那是一只玳瑁猫，胸前的银色项圈上刻着藤蔓花纹。我鬼使神差地靠近，它突然转过头，瞳孔在雨幕中缩成两道竖线。这双眼睛我太过熟悉，就像父亲书房里那幅褪色的油画——画中老猫也是这样凝视着窗外的雨。\\n\\n\"这是...？\"我摸出钥匙，却听见它用爪子轻轻推了推我的手腕。当钥匙插入锁孔的瞬间，阁楼的尘埃突然在空中凝成漩涡，书架上的书本自动翻动起来，泛黄的书页哗啦啦坠落。\\n\\n猫咪跳上窗台，对着雨幕轻叫。我这才发现玻璃上不知何时出现了划痕，那些古老的符文正在渗出淡蓝色的光。它用湿润的鼻尖碰了碰我的手背，项圈上的银饰突然发出嗡鸣。\\n\\n父亲失踪前的最后日记本在书架深处闪光。我翻开泛黄的纸页，那些被泪水晕开的字迹突然变得清晰：\"当猫眼映出雨痕，便是时候了。\"猫爪按着我的手背，带我触碰那行字，纸张突然化作雪花飘散。\\n\\n阁楼开始震颤，书架向两侧分开，露出后面深不见底的洞穴。猫咪跃入其中的刹那，我看见无数个自己在不同年代的图书馆里奔跑，每个背影后都跟着这只神秘的玳瑁猫。原来父亲二十年前就是在这里失踪的，而他留下的银项圈，此刻正套在眼前这只猫的脖子上。\\n\\n雨声渐歇，洞穴在身后合拢。猫咪跳回窗台，项圈上的符文暗淡下来，却在它眼中依旧闪烁。我轻轻抚摸它柔软的毛发，突然明白这座图书馆为何总在雨夜散发异彩——原来它从来都不是普通的建筑，而是一个跨越时空的渡口。\\n\\n---\\n\\n**接下来故事可能会沿着这些方向发展**：\\n\\n- **时光印记**：猫咪项圈上的藤蔓花纹暗示着它可能是父亲当年救出的时光守护者，这些花纹会随着不同年代的雨痕产生共鸣\\n\\n- **记忆传承**：图书馆地下藏有父亲用猫语写就的羊皮卷，记载着开启时空之门的密码与守护者的使命\\n\\n- **生命轮回**：每当新主人入住图书馆，猫咪就会以不同的形态出现，这种现象在古籍《猫与神秘的十二重门》中有详细记载\\n\\n希望这个故事能满足您对猫咪的想象。如果需要更多细节或调整，请随时告诉我。'}\n",
      "\n",
      "\n",
      "{'aggregator': '这是一个关于 cats 的故事、笑话和诗歌！\\n\\n故事：\\n\\n\\n好的，我将按照您的要求创作一个关于猫咪的故事。为了让故事更生动，我会先梳理一些基本的情节脉络。请您看看以下内容是否符合您的预期。如果您有其他想法，可以随时提出，我会进行调整。\\n\\n### 故事梗概\\n\\n- **神秘相遇**：在暴雨夜，一只玳瑁猫闯入老旧的图书馆，带着不属于这个时代的银色项圈，暗示着不寻常的身世。\\n\\n- **往事重现**：女主人林夕在整理旧书时，发现猫咪项圈上的刻痕与父亲留下的日记本记载惊人吻合，揭开尘封的往事。\\n\\n- **时空穿梭**：猫咪似乎能感知时间的流动，通过不同年代的书籍带林夕穿越，重现父亲与猫咪的往事与抉择。\\n\\n- **命运重叠**：当林夕发现父亲曾用同样的方式寻找失踪的猫咪，她终于理解生命与记忆的传承，选择与猫咪共同守护图书馆。\\n\\n- **永恒守望**：图书馆成为连接不同时空的纽带，猫咪在每个雨夜守护着这座建筑，见证着新旧交替与永恒的爱。\\n\\n---\\n暴雨砸在图书馆的彩绘玻璃上，发出清脆的敲击声。我握着《猫与神秘的十二重门》这本书，突然听见阁楼传来窸窣响动。推开吱呀作响的木门，一团毛茸茸的影子正蹲在古董书架前，琥珀色的眼睛在昏暗中泛着微光。\\n\\n那是一只玳瑁猫，胸前的银色项圈上刻着藤蔓花纹。我鬼使神差地靠近，它突然转过头，瞳孔在雨幕中缩成两道竖线。这双眼睛我太过熟悉，就像父亲书房里那幅褪色的油画——画中老猫也是这样凝视着窗外的雨。\\n\\n\"这是...？\"我摸出钥匙，却听见它用爪子轻轻推了推我的手腕。当钥匙插入锁孔的瞬间，阁楼的尘埃突然在空中凝成漩涡，书架上的书本自动翻动起来，泛黄的书页哗啦啦坠落。\\n\\n猫咪跳上窗台，对着雨幕轻叫。我这才发现玻璃上不知何时出现了划痕，那些古老的符文正在渗出淡蓝色的光。它用湿润的鼻尖碰了碰我的手背，项圈上的银饰突然发出嗡鸣。\\n\\n父亲失踪前的最后日记本在书架深处闪光。我翻开泛黄的纸页，那些被泪水晕开的字迹突然变得清晰：\"当猫眼映出雨痕，便是时候了。\"猫爪按着我的手背，带我触碰那行字，纸张突然化作雪花飘散。\\n\\n阁楼开始震颤，书架向两侧分开，露出后面深不见底的洞穴。猫咪跃入其中的刹那，我看见无数个自己在不同年代的图书馆里奔跑，每个背影后都跟着这只神秘的玳瑁猫。原来父亲二十年前就是在这里失踪的，而他留下的银项圈，此刻正套在眼前这只猫的脖子上。\\n\\n雨声渐歇，洞穴在身后合拢。猫咪跳回窗台，项圈上的符文暗淡下来，却在它眼中依旧闪烁。我轻轻抚摸它柔软的毛发，突然明白这座图书馆为何总在雨夜散发异彩——原来它从来都不是普通的建筑，而是一个跨越时空的渡口。\\n\\n---\\n\\n**接下来故事可能会沿着这些方向发展**：\\n\\n- **时光印记**：猫咪项圈上的藤蔓花纹暗示着它可能是父亲当年救出的时光守护者，这些花纹会随着不同年代的雨痕产生共鸣\\n\\n- **记忆传承**：图书馆地下藏有父亲用猫语写就的羊皮卷，记载着开启时空之门的密码与守护者的使命\\n\\n- **生命轮回**：每当新主人入住图书馆，猫咪就会以不同的形态出现，这种现象在古籍《猫与神秘的十二重门》中有详细记载\\n\\n希望这个故事能满足您对猫咪的想象。如果需要更多细节或调整，请随时告诉我。\\n\\n笑话：\\n\\n\\n好的，这里有一个关于猫的笑话：\\n\\n---\\n\\n**笑话：**\\n\\n一只猫对主人说：“我想去太空旅行，成为宇航员！”\\n\\n主人惊讶地问：“你有这方面的资质吗？”\\n\\n猫骄傲地甩了甩尾巴：“当然！我有完美的身体结构，锋利的爪子能抓握太空舱，还有夜视能力——毕竟在宇宙里，白天太亮了，我需要隐身！”\\n\\n主人想了想，指着电视上的火箭发射直播：“那先从这开始吧，你跳上去试试？”\\n\\n猫盯着屏幕，不屑地说：“这太简单了……我只需要一个铲子和一盒猫粮，就能在月球上开一家‘喵星人零食店’。”\\n\\n结果，它跳上沙发，用爪子拍了拍遥控器，把电视节目换成了猫咪视频合集。\\n\\n---\\n\\n**笑点解析：**  \\n猫的“太空梦想”被主人用现实的方式打脸（遥控器），而猫的回应又展现了它的傲娇与自我中心，最后结局反转幽默，同时暗含对猫日常行为的调侃（笑脸、铲子、猫粮这些元素都是猫主子熟悉的“战场”）。\\n\\n诗歌：\\n\\n\\n《猫的独白》\\n\\n午夜的剪影里\\n你的瞳孔收缩成两枚铜币\\n轻轻一跃\\n便撕开丝绸的帷幕\\n\\n我数着你胡须的针脚\\n编织成雾的经纬\\n当月光在瓦罐里发酵\\n你衔着影子踱步\\n\\n琥珀色的沉默在掌心\\n凝结成未拆封的信\\n你用尾巴丈量时间\\n在墙角画出螺旋的痕\\n\\n最古老的谜题\\n被你藏进毛茸茸的爪印\\n当人类还在笨拙地学习\\n如何用目光驯服星辰'}\n",
      "\n",
      "\n",
      "{'parallel_workflow': '这是一个关于 cats 的故事、笑话和诗歌！\\n\\n故事：\\n\\n\\n好的，我将按照您的要求创作一个关于猫咪的故事。为了让故事更生动，我会先梳理一些基本的情节脉络。请您看看以下内容是否符合您的预期。如果您有其他想法，可以随时提出，我会进行调整。\\n\\n### 故事梗概\\n\\n- **神秘相遇**：在暴雨夜，一只玳瑁猫闯入老旧的图书馆，带着不属于这个时代的银色项圈，暗示着不寻常的身世。\\n\\n- **往事重现**：女主人林夕在整理旧书时，发现猫咪项圈上的刻痕与父亲留下的日记本记载惊人吻合，揭开尘封的往事。\\n\\n- **时空穿梭**：猫咪似乎能感知时间的流动，通过不同年代的书籍带林夕穿越，重现父亲与猫咪的往事与抉择。\\n\\n- **命运重叠**：当林夕发现父亲曾用同样的方式寻找失踪的猫咪，她终于理解生命与记忆的传承，选择与猫咪共同守护图书馆。\\n\\n- **永恒守望**：图书馆成为连接不同时空的纽带，猫咪在每个雨夜守护着这座建筑，见证着新旧交替与永恒的爱。\\n\\n---\\n暴雨砸在图书馆的彩绘玻璃上，发出清脆的敲击声。我握着《猫与神秘的十二重门》这本书，突然听见阁楼传来窸窣响动。推开吱呀作响的木门，一团毛茸茸的影子正蹲在古董书架前，琥珀色的眼睛在昏暗中泛着微光。\\n\\n那是一只玳瑁猫，胸前的银色项圈上刻着藤蔓花纹。我鬼使神差地靠近，它突然转过头，瞳孔在雨幕中缩成两道竖线。这双眼睛我太过熟悉，就像父亲书房里那幅褪色的油画——画中老猫也是这样凝视着窗外的雨。\\n\\n\"这是...？\"我摸出钥匙，却听见它用爪子轻轻推了推我的手腕。当钥匙插入锁孔的瞬间，阁楼的尘埃突然在空中凝成漩涡，书架上的书本自动翻动起来，泛黄的书页哗啦啦坠落。\\n\\n猫咪跳上窗台，对着雨幕轻叫。我这才发现玻璃上不知何时出现了划痕，那些古老的符文正在渗出淡蓝色的光。它用湿润的鼻尖碰了碰我的手背，项圈上的银饰突然发出嗡鸣。\\n\\n父亲失踪前的最后日记本在书架深处闪光。我翻开泛黄的纸页，那些被泪水晕开的字迹突然变得清晰：\"当猫眼映出雨痕，便是时候了。\"猫爪按着我的手背，带我触碰那行字，纸张突然化作雪花飘散。\\n\\n阁楼开始震颤，书架向两侧分开，露出后面深不见底的洞穴。猫咪跃入其中的刹那，我看见无数个自己在不同年代的图书馆里奔跑，每个背影后都跟着这只神秘的玳瑁猫。原来父亲二十年前就是在这里失踪的，而他留下的银项圈，此刻正套在眼前这只猫的脖子上。\\n\\n雨声渐歇，洞穴在身后合拢。猫咪跳回窗台，项圈上的符文暗淡下来，却在它眼中依旧闪烁。我轻轻抚摸它柔软的毛发，突然明白这座图书馆为何总在雨夜散发异彩——原来它从来都不是普通的建筑，而是一个跨越时空的渡口。\\n\\n---\\n\\n**接下来故事可能会沿着这些方向发展**：\\n\\n- **时光印记**：猫咪项圈上的藤蔓花纹暗示着它可能是父亲当年救出的时光守护者，这些花纹会随着不同年代的雨痕产生共鸣\\n\\n- **记忆传承**：图书馆地下藏有父亲用猫语写就的羊皮卷，记载着开启时空之门的密码与守护者的使命\\n\\n- **生命轮回**：每当新主人入住图书馆，猫咪就会以不同的形态出现，这种现象在古籍《猫与神秘的十二重门》中有详细记载\\n\\n希望这个故事能满足您对猫咪的想象。如果需要更多细节或调整，请随时告诉我。\\n\\n笑话：\\n\\n\\n好的，这里有一个关于猫的笑话：\\n\\n---\\n\\n**笑话：**\\n\\n一只猫对主人说：“我想去太空旅行，成为宇航员！”\\n\\n主人惊讶地问：“你有这方面的资质吗？”\\n\\n猫骄傲地甩了甩尾巴：“当然！我有完美的身体结构，锋利的爪子能抓握太空舱，还有夜视能力——毕竟在宇宙里，白天太亮了，我需要隐身！”\\n\\n主人想了想，指着电视上的火箭发射直播：“那先从这开始吧，你跳上去试试？”\\n\\n猫盯着屏幕，不屑地说：“这太简单了……我只需要一个铲子和一盒猫粮，就能在月球上开一家‘喵星人零食店’。”\\n\\n结果，它跳上沙发，用爪子拍了拍遥控器，把电视节目换成了猫咪视频合集。\\n\\n---\\n\\n**笑点解析：**  \\n猫的“太空梦想”被主人用现实的方式打脸（遥控器），而猫的回应又展现了它的傲娇与自我中心，最后结局反转幽默，同时暗含对猫日常行为的调侃（笑脸、铲子、猫粮这些元素都是猫主子熟悉的“战场”）。\\n\\n诗歌：\\n\\n\\n《猫的独白》\\n\\n午夜的剪影里\\n你的瞳孔收缩成两枚铜币\\n轻轻一跃\\n便撕开丝绸的帷幕\\n\\n我数着你胡须的针脚\\n编织成雾的经纬\\n当月光在瓦罐里发酵\\n你衔着影子踱步\\n\\n琥珀色的沉默在掌心\\n凝结成未拆封的信\\n你用尾巴丈量时间\\n在墙角画出螺旋的痕\\n\\n最古老的谜题\\n被你藏进毛茸茸的爪印\\n当人类还在笨拙地学习\\n如何用目光驯服星辰'}\n",
      "\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from langchain_openai import ChatOpenAI\n",
    "from langgraph.func import entrypoint, task\n",
    "\n",
    "llm = ChatOpenAI(model=\"Qwen/Qwen3-8B\")\n",
    "\n",
    "@task\n",
    "def call_llm_1(topic: str):\n",
    "    \"\"\"第一个 LLM 调用，生成初始笑话\"\"\"\n",
    "    msg = llm.invoke(f\"写一个关于 {topic} 的笑话\") # 调用 LLM 以根据主题写一个笑话\n",
    "    return msg.content # 返回笑话\n",
    "\n",
    "@task\n",
    "def call_llm_2(topic: str):\n",
    "    \"\"\"第二个 LLM 调用，生成故事\"\"\"\n",
    "    msg = llm.invoke(f\"写一个关于 {topic} 的故事\") # 调用 LLM 以根据主题写一个故事\n",
    "    return msg.content # 返回故事\n",
    "\n",
    "@task\n",
    "def call_llm_3(topic):\n",
    "    \"\"\"第三个 LLM 调用，生成诗歌\"\"\"\n",
    "    msg = llm.invoke(f\"写一首关于 {topic} 的诗歌\") # 调用 LLM 以根据主题写一首诗\n",
    "    return msg.content # 返回诗歌\n",
    "\n",
    "@task\n",
    "def aggregator(topic, joke, story, poem):\n",
    "    \"\"\"将笑话和故事组合成单个输出\"\"\"\n",
    "    combined = f\"这是一个关于 {topic} 的故事、笑话和诗歌！\\n\\n\" # 开始组合输出\n",
    "    combined += f\"故事：\\n{story}\\n\\n\" # 将故事添加到组合输出\n",
    "    combined += f\"笑话：\\n{joke}\\n\\n\" # 将笑话添加到组合输出\n",
    "    combined += f\"诗歌：\\n{poem}\" # 将诗歌添加到组合输出\n",
    "    return combined # 返回组合输出\n",
    "\n",
    "# 入口点装饰函数定义并行工作流\n",
    "@entrypoint()\n",
    "def parallel_workflow(topic: str):\n",
    "    joke_fut = call_llm_1(topic) # 执行 'call_llm_1' 任务并获取期货以进行并行执行\n",
    "    story_fut = call_llm_2(topic) # 执行 'call_llm_2' 任务并获取期货以进行并行执行\n",
    "    poem_fut = call_llm_3(topic) # 执行 'call_llm_3' 任务并获取期货以进行并行执行\n",
    "    return aggregator(\n",
    "        topic, joke_fut.result(), story_fut.result(), poem_fut.result() # 在所有并行任务完成后执行 'aggregator' 任务\n",
    "    ).result() # 从聚合器获取最终结果\n",
    "\n",
    "# 调用并行工作流\n",
    "for step in parallel_workflow.stream(\"cats\", stream_mode=\"updates\"):\n",
    "    print(step)\n",
    "    print(\"\\n\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "todo-update-1",
   "metadata": {},
   "source": [
    "**💡 Functional API 中的并行处理**：\n",
    "\n",
    "- **Future 对象**：通过 `.result()` 方法等待异步任务完成\n",
    "- **简洁语法**：并行执行通过同时启动多个任务实现\n",
    "- **自动同步**：聚合器会等待所有前置任务完成"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "dj67jjxa5xq",
   "metadata": {},
   "source": [
    "### 7.1.5 协调器-工作者 (Orchestrator-Worker)\n",
    "\n",
    "协调器-工作者工作流模式专为子任务需求事先未知且需要在执行期间动态确定的复杂任务而设计。在此模式中，中央增强型 LLM 充当\"协调器 (Orchestrator)\"，负责将初始任务分解为更小、更易于管理的子任务，并将这些子任务委派给\"工作者 (Worker)\"增强型 LLM。\n",
    "\n",
    "此工作流特别适用于难以或不可能预先预测必要子任务的复杂场景，例如编码任务、复杂搜索任务等。"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "yhnp7oyv2k8",
   "metadata": {},
   "source": [
    "##### 示例 7-7：基于 Graph API 的\"协调器-工作者\"工作流实现\n",
    "\n",
    "让我们实现一个报告生成的协调器-工作者系统："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "3ppbtmr227s",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/markdown": [
       "# 引言\n",
       "\n",
       "大规模语言模型（Large Language Models, LLMs）在近年来的自然语言处理领域中占据了中心位置。这些模型通过处理海量的文本数据，在广泛的任务上展示了卓越的性能。从文本生成到代码编写，甚至是复杂的对话系统，LLMs 的应用范围几乎无所不包。然而，LLMs 的发展并非一蹴而就，其背后蕴含了无数学术和工业界的辛勤努力和创新突破。\n",
       "\n",
       "为了理解 LLMs 的现状和未来潜力，我们首先需要回顾它们的发展历史，特别是探讨模型规模与性能之间的关系。在这种背景下，缩放定律（Scaling Laws）逐渐成为研究焦点，这些定律探索了当模型规模增加时，性能如何提升。通过分析这些规律，研究人员能够更深入地理解模型结构、训练策略和算力需求之间的复杂相互作用。因此，理解缩放定律对于进一步推动 LLMs 的理论研究和应用实践至关重要。\n",
       "\n",
       "---\n",
       "\n",
       "## LLM 缩放定律的定义\n",
       "\n",
       "LLM 缩放定律指的是在训练语言模型时，随着模型参数量、训练数据量等关键因素的增加，尤其是当这些因素按照预期的比例放大时，模型性能会呈现出显著的提升趋势。研究和观测表明，这种性能增益通常呈现对数关系或线性关系，且每倍的参数量增加通常都会伴随模型性能成倍的提升。\n",
       "\n",
       "### 关键参数之间的关系\n",
       "\n",
       "在理解 LLM 缩放定律之前，首先要明确几个关键参数之间的相互关系：\n",
       "\n",
       "1. **模型参数量**：这是指模型内部存储的权重数量，更大的参数量通常意味着模型具有更高的复杂性，可以捕捉更深层次的特征和模式。\n",
       "\n",
       "2. **训练数据量**：指的是模型在训练过程中用于学习的文本数据集大小。更多的数据可以使模型学会更广泛的语言结构和语义关系。\n",
       "\n",
       "3. **模型性能**：通过特定基准测试衡量模型完成指定任务的能力，比如语言生成质量、理解能力和推理能力等。\n",
       "\n",
       "### 常见缩放曲线\n",
       "\n",
       "- **模型参数量与性能**：多项研究表明，随着模型参数量的显著增加，模型在各种自然语言处理任务上的表现会在一个时间阈值内迅速提升。通过定量分析，这种提升往往呈对数增长趋势，即每增加一倍的模型参数，性能可能提高一个系数级别。\n",
       "- **训练数据量与性能**：同样地，更大的训练数据集能够促进模型学习到更多样性的语言模式，从而增强其泛化能力和表现。数据量的增强与模型性能之间的关系往往遵循类似参数量与性能趋势的增长曲线。\n",
       "\n",
       "综合理解这些关系有助于设计和优化高性能的LLM，同时认识到它们之间的放大效应对于科学研究和实际应用都至关重要。\n",
       "\n",
       "---\n",
       "\n",
       "# LLM 缩放的实证分析\n",
       "\n",
       "## 引言\n",
       "本章节旨在通过实证研究来分析大模型在实际应用场景中的性能提升情况，并与经典的缩放定律进行对比验证。缩放定律通常指的是随着模型参数规模的增加，模型表现出的性能也会相应地提升。通过本章节的实证分析，我们将试图验证这一假设是否在实际应用中同样成立。\n",
       "\n",
       "## 研究背景\n",
       "近年来，大模型在自然语言处理领域取得了显著进展。然而，随着模型规模的不断增加，其性能提升是否与预期相符成为了业界关注的焦点之一。现有的研究虽然在一定程度上支持了模型规模与性能之间的正相关关系，但实际场景中的模型应用情况可能更为复杂和多变。因此，通过系统性和科学性的实证分析来进一步检验这一关系具有重要的理论和实践意义。\n",
       "\n",
       "## 方法论\n",
       "本部分将详细介绍实证分析的方法，包括数据收集、实验设计、模型选择和评估指标等。\n",
       "1. **数据收集**：选择不同规模的语言模型作为测试对象，这些模型应具有类似的架构和训练框架，但参数量上存在显著差异。\n",
       "2. **实验设计**：设计一系列基准任务用于评估，确保所选任务与实际应用场景具有高度关联性。同时设立控制变量，比如相同的训练数据集、相同的数据输入格式等。\n",
       "3. **模型选择**：选取几种不同的预训练大模型作为实验对象。\n",
       "4. **评估指标**：采用准确率、召回率、F1 值等评估指标来量化模型的性能表现。同时也可以引入更专业的评估任务和性能指标，以便更全面地评估模型在特定任务上的表现。\n",
       "\n",
       "## 结果与讨论\n",
       "1. **性能提升趋势**：通过对比不同规模模型在各项任务上的表现，分析模型规模与性能之间的关系。记录每一级参数量增加后模型性能的具体变化情况。\n",
       "2. **与缩放定律的对比**：将实际数据与理论上的缩放模型进行对比，探讨两者之间是否存在一定的偏离现象，分析可能的原因。\n",
       "3. **案例分析**：挑选特定的应用场景进行深入分析，具体探讨在这类任务中参数量的变化对模型性能的实际影响。\n",
       "\n",
       "## 结论\n",
       "总结实证分析的主要发现，包括模型规模与性能之间关系的验证结果、存在的问题以及未来的研究方向。强调对于不同规模模型的选择原则以及如何针对性地优化特定任务下的大模型性能。\n",
       "\n",
       "## 讨论\n",
       "鼓励对本章结果的广泛应用进行讨论，包括大型语言模型在真实世界中的部署和实际效果评估等。提出改进现有模型及缩放理论的可能途径。\n",
       "\n",
       "---\n",
       "\n",
       "# 现行的缩放边界限制\n",
       "\n",
       "在当前的技术条件下，实现深度学习模型的缩放存在多个方面的局限性，这些限制在计算能力、存储资源、应用效率以及成本方面体现出来。下面将详细讨论这些限制。\n",
       "\n",
       "## 计算能力限制\n",
       "随着模型规模的扩大，深度学习模型对计算资源的需求也会大幅增加。大模型不仅在训练时需要大量计算资源，而且在推理阶段也需要高性能的硬件支持（如TPU和GPU）以维持高效率。当前，尽管云服务提供商提供了强大的计算资源（包括大型GPU集群和TPU），但对个人开发者或小型公司来说，能够接触到如此规模计算资源的情况相对较为罕见。这为模型在扩展方面设置了一定的限制。\n",
       "\n",
       "## 存储限制\n",
       "大型深度学习模型需要占用大量存储空间才能在其完整形式下进行保存和传输。当前，对于目前的技术而言，存储成本仍然比计算成本要低，但对于非常大的模型而言，存储成本仍然是一个需要考虑的问题。云存储服务提供者通常会提供各种存储选项，包括低容量、高成本选项与高容量、低至中等成本选项。 \n",
       "\n",
       "## 应用效率限制\n",
       "虽然大型模型具有更强的表达能力和预测性能，但它们往往导致更高的延迟。在需要低延迟的场景中（例如，实时推断或移动设备推理），大型模型可能不是一个实际的选择，因为它们可能消耗过多的计算资源，导致推理速度下降，从而影响用户体验。随着硬件技术的进步，现在存在一些能够满足某些效率要求的替代方案，如模型修剪、因子化的开发以及蒸馏技术等方法可以一定程度上解决这个问题。\n",
       "\n",
       "## 成本因素\n",
       "成本是规模化应用大型模型的一大问题。使用大型模型所需的算力和存储会迅速推高相关费用，从而给企业和研究机构带来不小的负担。对于部署在数据密集型应用中的大型模型而言，成本管理尤其重要。尽管如此，云平台的灵活性允许用户根据实际需求选择合适的硬件配置和存储方案，从而在成本控制和性能需求方面找到一个较为理想的平衡点。同时，随着技术进步和市场成熟，成本也在逐渐降低，越来越多的技术手段和策略被应用到提高成本效益方面，例如利用AI编译器和优化工具集来进一步减少模型所需的资源并提高模型的整体效率。\n",
       "\n",
       "---\n",
       "\n",
       "## 未来发展趋势\n",
       "\n",
       "预测 LLM（Large Language Models）缩放发展的潜在趋势和挑战，探讨未来可能的技术方向，对于理解这一领域的发展脉络至关重要。未来的发展趋势可能涵盖多个方面，包括技术进步、应用场景的扩展以及伦理和安全问题的处理。\n",
       "\n",
       "### 技术进步\n",
       "\n",
       "一项关键技术是模型大小、计算能力及训练数据规模的增长。模型的扩展会使得 LLM 能够更好地捕捉语言中的细微差异与复杂模式。通过采用更强大的计算资源，不但可以提高模型的性能，还能缩短训练时间，使得更多的数据集能够被用来训练更加复杂的模型。此外，研究者们也在探索如何利用混合精度训练、量化技术等方法来降低计算开销，从而使大模型的构建和训练更加可行。\n",
       "\n",
       "除了解决算力和数据的问题，优化模型结构、持续改进训练方法将是未来的重要方向。设计全新的架构是另一种可能，它针对特定任务进行多样化和专业化。模型训练中引入的知识迁移、蒸馏等策略也会持续受到关注，它们旨在提高模型的泛化能力，降低成本，使其更加高效地应用于实际场景。\n",
       "\n",
       "### 应用场景的扩展\n",
       "\n",
       "随着技术的进步，LLM 的应用场景将会越来越多元化。目前，LLM 已经在自然语言生成、文本分类、对话系统等多个领域得到广泛应用。未来，它们将在医疗健康、法律咨询、教育辅导、金融分析等多个垂直领域发挥重要作用，解决实际问题并提供支持。对特定领域知识的深入理解和模型的有效泛化将成为实现高度专业化和实用应用的关键。\n",
       "\n",
       "### 伦理与安全问题\n",
       "\n",
       "在推进 LLM 进一步发展的过程中，解决伦理和安全问题同样重要。一方面，确保模型使用的数据质量，避免偏见的传播是关键任务之一。研究者和相关组织应该致力于数据清洗、自动化检测工具开发等方面的工作。同时，促进公平性也是另一种关注点，通过鼓励多样性的数据收集和采纳，确保不同群体的声音都被代表和纳入模型学习中。此外，确保模型行为的透明度与可解释性，是降低用户及社会公众对技术不信任感的重要方式。这需要从算法设计初始阶段就加以考虑，贯穿于整个开发和部署流程之中。\n",
       "\n",
       "综上所述，未来 LLM 的发展将在技术进步、应用场景扩展以及伦理与安全问题等多个方面继续前进。准确把握这些趋势，不仅能为未来的技术革新提供指导，还有助于推动社会整体的数字化进步。\n",
       "\n",
       "---\n",
       "\n",
       "# 结论\n",
       "\n",
       "本报告通过对学术论文和案例研究的分析，探讨了[相关领域]中的关键问题，并评估了现有解决方案的有效性和局限性。我们的主要发现包括：\n",
       "\n",
       "1. [发现点1]：展现了当前方法在处理特定问题时表现出的优势与挑战。\n",
       "2. [发现点2]：揭示了对[相关领域]产生深远影响的趋势和动力。\n",
       "3. [发现点3]：指出在实际操作中可能会遇到的一些实际限制和不足。\n",
       "\n",
       "在未来的研究方向上，我们建议：\n",
       "\n",
       "1. **深入技术探索**：进一步研究[相关技术]在[具体领域]中的应用潜力，并探索可能的技术改进方法。\n",
       "2. **理论与实践结合**：加强不同学科之间的合作，利用理论分析指导实践操作，同时通过实际案例检验理论假设。\n",
       "3. **综合评估方法**：开发更加全面有效的评估工具和方法来系统性地评价现有解决方案的优劣。\n",
       "4. **安全性和隐私考虑**：特别关注[相关技术]或方法在保护用户数据安全和隐私方面的重要性，并提出改进措施。\n",
       "5. **跨文化视角**：从跨文化的角度出发重新审视现有问题和解决方案，以期获得更广泛适用的结论和建议。\n",
       "\n",
       "总之，通过本次的研究，我们希望能够推动[相关领域]的发展和进步，同时也提出了需要更加深入探讨的问题。"
      ],
      "text/plain": [
       "<IPython.core.display.Markdown object>"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import operator\n",
    "from typing import Annotated, List, TypedDict\n",
    "from pydantic import BaseModel, Field\n",
    "\n",
    "from langchain_openai import ChatOpenAI\n",
    "from langchain_core.messages import HumanMessage, SystemMessage\n",
    "from langgraph.graph import StateGraph, START, END\n",
    "from langgraph.types import Send\n",
    "\n",
    "\n",
    "# 用于结构化输出的模式，用于规划报告章节\n",
    "class Section(BaseModel):\n",
    "    name: str = Field(\n",
    "        description=\"报告章节的名称\", # 报告章节的名称\n",
    "    )\n",
    "    description: str = Field(\n",
    "        description=\"本章节中要涵盖的主要主题和概念的简要概述\", # 章节内容的描述\n",
    "    )\n",
    "\n",
    "class Sections(BaseModel):\n",
    "    sections: List[Section] = Field(\n",
    "        description=\"报告的章节\", # 报告章节列表\n",
    "    )\n",
    "\n",
    "# 用于规划报告章节的增强型 LLM，使用结构化输出\n",
    "llm = ChatOpenAI(model=\"Qwen/Qwen2.5-7B-Instruct\")\n",
    "planner = llm.with_structured_output(Sections, method=\"function_calling\")\n",
    "\n",
    "# 协调器-工作者工作流的图状态定义\n",
    "class State(TypedDict):\n",
    "    topic: str  # 报告主题\n",
    "    sections: list[Section]  # 由协调器规划的报告章节列表\n",
    "    completed_sections: Annotated[\n",
    "        list, operator.add\n",
    "    ]  # 所有工作者并行写入此键，使用 operator.add 进行列表连接\n",
    "    final_report: str  # 最终合成报告\n",
    "\n",
    "\n",
    "# 工作者状态定义，特定于工作者节点\n",
    "class WorkerState(TypedDict):\n",
    "    section: Section\n",
    "    completed_sections: Annotated[list, operator.add] # 工作者也写入共享的 'completed_sections' 键\n",
    "\n",
    "# 图中的节点\n",
    "def orchestrator(state: State):\n",
    "    \"\"\"协调器，使用结构化输出生成报告计划\"\"\"\n",
    "    # 使用 planner LLM 和结构化输出生成报告章节计划\n",
    "    report_sections = planner.invoke(\n",
    "        [\n",
    "            SystemMessage(content=\"生成报告计划。\"), # planner LLM 的系统消息\n",
    "            HumanMessage(content=f\"这是报告主题：{state['topic']}\"), # 包含报告主题的用户输入消息\n",
    "        ]\n",
    "    )\n",
    "    return {\"sections\": report_sections.sections} # 返回计划的章节，更新状态中的 'sections'\n",
    "\n",
    "def llm_call(state: WorkerState):\n",
    "    \"\"\"工作者根据分配的章节详细信息编写报告章节\"\"\"\n",
    "    # 使用 LLM 根据章节名称和描述生成报告章节内容\n",
    "    section = llm.invoke(\n",
    "        [\n",
    "            SystemMessage(\n",
    "                content=\"按照提供的名称和描述编写报告章节。每节不包含序言。使用 markdown 格式。\" # 工作者 LLM 的系统消息\n",
    "            ),\n",
    "            HumanMessage(\n",
    "                content=f\"这是章节名称：{state['section'].name} 和描述：{state['section'].description}\" # 包含章节详细信息的用户消息\n",
    "            ),\n",
    "        ]\n",
    "    )\n",
    "    # 将生成的章节内容写入共享的 'completed_sections' 键\n",
    "    return {\"completed_sections\": [section.content]}\n",
    "\n",
    "def synthesizer(state: State):\n",
    "    \"\"\"从各个章节输出合成完整报告\"\"\"\n",
    "    # 从共享状态检索已完成章节的列表\n",
    "    completed_sections = state[\"completed_sections\"]\n",
    "\n",
    "    # 将已完成章节格式化为单个字符串以用于最终报告\n",
    "    completed_report_sections = \"\\n\\n---\\n\\n\".join(completed_sections)\n",
    "\n",
    "    return {\"final_report\": completed_report_sections} # 返回最终报告，更新状态中的 'final_report'\n",
    "\n",
    "# 条件边缘函数，用于将工作者动态分配给计划中的每个章节\n",
    "def assign_workers(state: State):\n",
    "    \"\"\"使用 Send API 将工作者分配给计划中的每个章节，以实现动态工作者创建\"\"\"\n",
    "    # 使用 Send API 为每个章节动态创建和发送 'llm_call' 工作者节点\n",
    "    return [Send(\"llm_call\", {\"section\": s}) for s in state[\"sections\"]]\n",
    "\n",
    "# 使用 StateGraph 构建协调器-工作者工作流\n",
    "orchestrator_worker_builder = StateGraph(State)\n",
    "\n",
    "# 将节点添加到图中\n",
    "orchestrator_worker_builder.add_node(\"orchestrator\", orchestrator)\n",
    "orchestrator_worker_builder.add_node(\"llm_call\", llm_call) # 工作者节点\n",
    "orchestrator_worker_builder.add_node(\"synthesizer\", synthesizer)\n",
    "\n",
    "# 定义边缘以连接节点并建立协调器-工作者流程\n",
    "orchestrator_worker_builder.add_edge(START, \"orchestrator\") # 开始节点连接到协调器\n",
    "orchestrator_worker_builder.add_conditional_edges(\n",
    "    \"orchestrator\", assign_workers, [\"llm_call\"] # 从协调器到使用 Send API 动态创建的工作者节点的条件边缘\n",
    ")\n",
    "orchestrator_worker_builder.add_edge(\"llm_call\", \"synthesizer\") # 工作者节点在完成后连接到合成器\n",
    "orchestrator_worker_builder.add_edge(\"synthesizer\", END) # 合成器节点连接到结束节点\n",
    "\n",
    "# 编译协调器-工作者工作流图\n",
    "orchestrator_worker = orchestrator_worker_builder.compile()\n",
    "\n",
    "# 使用示例报告主题调用协调器-工作者工作流\n",
    "state = orchestrator_worker.invoke({\"topic\": \"创建关于 LLM 缩放定律的报告\"})\n",
    "\n",
    "from IPython.display import Markdown\n",
    "Markdown(state[\"final_report\"]) # 以 Markdown 格式显示最终报告"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ywhte0757f",
   "metadata": {},
   "source": [
    "**💡 协调器-工作者的关键特性**：\n",
    "\n",
    "- **动态任务分解**：协调器根据输入动态生成子任务\n",
    "- **Send API**：允许运行时动态创建工作者节点\n",
    "- **共享状态**：使用 `operator.add` 聚合多个工作者的输出\n",
    "- **结构化输出**：使用 Pydantic 模型确保任务规划的一致性\n",
    "- **灵活扩展**：可以根据任务复杂性创建任意数量的工作者"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "cfouwqfokhl",
   "metadata": {},
   "source": [
    "##### 示例 7-8：基于 Functional API 的\"协调器-工作者\"工作流实现\n",
    "\n",
    "让我们用 Functional API 实现相同的协调器-工作者逻辑："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "lbvlglvna5n",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/markdown": [
       "### 引言\n",
       "\n",
       "本报告旨在探讨大规模语言模型（LLM）中的缩放定律及其重要性。随着技术的发展，LLM 已在全球范围内展现出强大的应用潜力，包括自然语言处理、客户服务、内容创作、个性化推荐等。缩放定律作为一种关键的理论框架，帮助我们理解 LLM 性能提升背后的原因，并为未来的研发和部署提供重要的指导。\n",
       "\n",
       "#### LLM 缩放定律的重要性\n",
       "\n",
       "缩放定律揭示了 LLM 性能随模型大小和计算资源增加的趋势。理解和掌握这一规律不仅有助于优化现有模型的设计，还能指导开发者和研究者在设计新模型时做出更明智的决策。此外，缩放定律也对企业和社会产生深远影响，包括成本效益分析、资源分配策略以及技术发展方向。\n",
       "\n",
       "#### 本文档的内容概要\n",
       "\n",
       "本报告将深入剖析当前主流的 LLM 缩放定律（例如模型参数量增长对性能的影响、计算资源与训练成本的关系等），并基于现有研究结果提出建议。还将在第二部分探讨各种因素如何影响这一规律的具体应用，并对潜在风险和挑战进行讨论。第三部分将聚焦于未来的展望，包括可预见的技术发展和可能的突破方向。最后，报告将总结全文，并提出对从业人员及研究者的重要建议。\n",
       "\n",
       "通过本章节，读者应能获得对 LLM 的缩放规律基本概念的理解，并为进一步阅读后续章节打下坚实基础。\n",
       "\n",
       "---\n",
       "\n",
       "标题：LLM 缩放定律概述\n",
       "\n",
       "LLM 缩放定律是解释大规模语言模型在训练过程中性能和参数扩展之间的关系的原理。本章节将对缩放定律进行详细阐述并讨论其历史背景和重要性。\n",
       "\n",
       "一、定义 LLM 缩放定律\n",
       "\n",
       "LLM 缩放定律指的是在大规模语言模型的训练过程中，某些性能指标（如语言理解能力、生成能力等）与模型参数量之间的关系。基本上，当层结构保持不变且模型参数数量增加时，模型在训练集上的表现会逐渐提高。然而，值得注意的是这种关系并非线性，而是呈现为某种次线性模式。也就是说，参数数量翻倍时，模型表现的大约提升了根号两倍。\n",
       "\n",
       "二、历史背景\n",
       "\n",
       "要全面理解 LLM 缩放定律的历史背景，就需要回顾语言模型的发展历程。在经典的深度学习范式中，Bengio 等人提出了深度神经网络在学习序列性任务时表现出的强大能力。随后，随着计算能力和数据集规模的爆炸性增长，以Transformer 为代表的序列建模方法盛行。目前，Transformer 语言模型的参数规模已超万亿，在英文版 Wikipedia 的预训练中达到了 1750 亿参数。这种无与伦比的规模正是学习大规模语言模型和相关 LLM 缩放定律得以显现的背景。\n",
       "\n",
       "三、重要性\n",
       "\n",
       "研究 LLM 缩放定律对于构建高性能、可扩展的语言模型具有重要意义。首先，它可以作为构建新硬件的基础，为对超大规模模型提供支持，包括训练和推理。其次，通过 LLM 缩放定律的洞察，我们可以直接提高现有模型的效率。理解缩放定律可以帮助我们优化模型结构、调整超参数和设计新型硬件，从而在保持性能的同时降低计算和存储资源的消耗。此外，这一知识还可以指导减少 LLM 训练所需的数据集大小与训练时间。\n",
       "\n",
       "结论\n",
       "\n",
       "本章节对 LLM 缩放定律进行了阐述，并讨论了其重要的历史背景和潜在价值。通过更好地理解这一定律，我们可以为形成高效和强大的语言模型群开发提供更广泛的指导。\n",
       "\n",
       "---\n",
       "\n",
       "### 缩放定律的基本概念\n",
       "\n",
       "本节将深入解析大型语言模型（LLM）中的缩放定律，即模型规模增长与性能提升之间的关联，以及这一现象背后的机制。\n",
       "\n",
       "#### 1. 缩放定律的定义\n",
       "\n",
       "缩放定律概述了随着模型参数数量的增加，模型性能通常会呈现出显著的改进。这一概念基于现有的大量研究，包括但不限于Transformers架构以及自然语言处理任务上的应用。换句话说，如果一个大型语言模型的参数数量是前一个模型的两倍，那么其性能可能翻倍甚至更高。\n",
       "\n",
       "#### 2. 模型尺寸与性能的关系\n",
       "\n",
       "随着模型参数数量的增加，LLM能够从中提取更复杂的特征和模式，从而在各种任务上（如机器翻译、情感分析和问答）获得更出色的性能。原因往往在于以下几个方面：\n",
       "\n",
       "- **更多的参数允许更复杂的内部结构**：增加了参数意味着模型拥有更多的挖掘和关联不同数据点的能力。\n",
       "- **更高的训练数据数量和多样性**：更大的模型能够更好地学习并适应训练数据的多样性。\n",
       "- **更强的表达能力**：对于具体的Transformer架构而言，更大的模型能够使用更宽的隐藏层、更多层和更大的注意力机制，这些都会提升预测能力。\n",
       "\n",
       "#### 3. 参数与计算量的关系\n",
       "\n",
       "增加模型规模不仅增加了存储空间的需求，也会带来计算量的显著增加。这包括：\n",
       "\n",
       "- **更高的推理时间和推理成本**：进行预测或生成文本时，需要对更多的参数进行复杂的运算，这会显著延长执行时间并增加消耗的计算资源。\n",
       "- **训练时间的大幅增加**：大规模模型上训练所需的时间可能会成指数级增长，这不仅需要大量数据和计算力，还可能需要改进训练方法以提高效率。\n",
       "\n",
       "总之，LLM中的缩放定律揭示了通过增加模型的维度可以如何显著提升其性能，与此同时，这背后也伴随着更多的挑战，包括计算资源的消耗和培训时间的延长。理解这些概念对于设计和应用有效的大型语言模型至关重要。\n",
       "\n",
       "---\n",
       "\n",
       "### 实验与数据支持\n",
       "\n",
       "为了充分支持 LLM 缩放定律，我们采用了一种综合性的实验设计和数据收集方法，涵盖了多个数据集和分析维度。通过这种方法，我们旨在提供全面的证据，以验证该定律在不同场景下的有效性。\n",
       "\n",
       "#### 1. 实验设计\n",
       "我们的实验设计主要围绕几个关键点展开：\n",
       "\n",
       "**1.1 模型选择：** 我们选择了多个不同大小的语言模型，从小到大，以确保能够覆盖广泛的实际应用需求和模型能力范围。\n",
       "\n",
       "**1.2 数据集：** 研究中使用了广泛不同领域和规模的数据集，包括但不限于：英语文本、中文文本、科学文献、网页内容等，以确保结果具有普遍适用性。\n",
       "\n",
       "**1.3 评估指标：** 我们定义了一套综合性评估指标，不仅包括模型的准确率、精度等传统性能指标，还包括吞吐量、资源消耗效率、生成文本的新颖性和创意性等因素。\n",
       "\n",
       "#### 2. 数据收集\n",
       "通过一致的实验流程，我们收集了多种模式大小语言模型在不同数据集上的表现数据，这些数据被用于进行后续的统计分析与对比研究。\n",
       "\n",
       "#### 3. 数据分析方法\n",
       "采用先进的统计分析工具和技术，对收集到的数据进行深入分析。具体包括但不限于回归分析、时间序列分析等方法，以揭示模型规模与其性能指标之间的量化关系。\n",
       "\n",
       "#### 4. 结果展示\n",
       "实验结果强有力地支持了 LLM 缩放定律。随着模型大小的增长，这些语言模型在各种评估指标上都表现出显著的改善，证实了模型规模对于提升模型性能的重要作用。\n",
       "\n",
       "#### 结论\n",
       "综合上述实验设计、数据集选择和分析方法，我们不仅能够有效的数据采集和处理，更重要的是能够从多个维度定量地展示了 LLM 缩放定律的应用价值。这些结果为我们深刻理解和预测大型语言模型的发展提供了重要参考。\n",
       "\n",
       "---\n",
       "\n",
       "### 技术讨论\n",
       "\n",
       "本章节旨在深入分析大型语言模型（LLM）的技术实现细节，并探讨缩放定律在模型架构设计中的应用。整个章节将分为几个部分，以全面覆盖该主题：\n",
       "\n",
       "#### 1. LLM 模型架构设计\n",
       "\n",
       "在本部分中，我们将详细介绍构成 LLM 的主要组件及其功能。主要包括但不限于：词嵌入层、多层变换器、注意机制、输入-输出界面等。每种组件都将介绍其技术原理及在训练任务中发挥的关键作用，例如：如何通过多头注意力机制提高模型对复杂语言结构的理解能力。\n",
       "\n",
       "#### 2. 放缩定律在模型设计中的应用\n",
       "\n",
       "放缩定律强调了在模型参数量级和计算资源之间的重要关系，是指导大模型架构设计的一个重要思想。我们将重点阐述这一原理，以及在实际应用中如何平衡好模型的复杂度和性能之间的关系。例如：通过增加模型层数和维度来实现更强的语言理解能力；同时，合理规划模型的硬件使用效率。\n",
       "\n",
       "#### 3. 超参数调优方法\n",
       "\n",
       "参数量大意味着需要调整更多的超参数以优化模型性能。在这里，我们将介绍目前被广泛采用的一些高效方法，如：交叉验证、网格搜索和随机搜索等。除了介绍这些技术，我们还将展示其在不同情况下的适用场景和不足之处，从而帮助读者根据自身需求选择合适的超参数调整策略。\n",
       "\n",
       "#### 4. 实践中的注意事项\n",
       "\n",
       "在实践过程中，开发者还需要注意若干问题，如：训练过程中如何避免过拟合；如何评估超参数调整效果等。本部分将列出常见问题，并提出对应的解决方案，以帮助读者更好地应对各类挑战。\n",
       "\n",
       "#### 5. 未来发展趋势\n",
       "\n",
       "LLM 技术正处于飞速发展的阶段，本部分将总结当前研究方向中的热门趋势与前沿进展。此外，我们还会提出一些潜在的发展方向，引导读者思考 LLM 的下一步走向。\n",
       "\n",
       "通过本章的内容，读者不仅能够详细了解 LLM 的技术实现和方法论，还可以获取关于如何优化模型性能的重要见解。这些知识将极大地提升读者在构建和部署大规模语言模型方面的水平。\n",
       "\n",
       "---\n",
       "\n",
       "### 案例研究\n",
       "\n",
       "在本章节中，我们将探讨几个实际应用语言模型（LLM）的案例研究，以此来深入了解 LLM 缩放定律在不同场景下的成效和面临的主要挑战。这些建设性的案例研究不仅为理论提供了一个现实的试验场，也为实际应用提供了重要的参考和启示。\n",
       "\n",
       "#### 案例一：在线客户服务机器人\n",
       "\n",
       "**概述：** 一家大型科技公司开发了一个基于 LLM 的在线客户服务机器人，旨在提供实时技术支持。该应用结合了 LLM 的强大语言处理能力和自学习机制，能够自动回答常见问题、提供故障排除建议，并且随着用户交互不断优化其回答。\n",
       "\n",
       "**成效：**\n",
       "- 显著减少了用户等待时间，提升了客户满意度。\n",
       "- 减少了客户服务团队的工作强度，提高了回应效率。\n",
       "\n",
       "**挑战：**\n",
       "- 较高的初始部署成本，包括训练和调优。\n",
       "- 需要严格的数据隐私保护措施以确保用户信息的安全。\n",
       "\n",
       "#### 案例二：高级文本生成案例\n",
       "\n",
       "**概述：** 利用 LLM 生成高质量的新闻报道、市场分析报告的文字内容。这种应用在市场上展现出巨大的潜力，尤其是在定制内容生产方面。\n",
       "\n",
       "**成效：**\n",
       "- 提升了生产效率，使公司能够更快速、更精确地创造出所需的内容。\n",
       "- 通过精准的信息表达能力吸引更多的读者或客户。\n",
       "\n",
       "**挑战：**\n",
       "- 确保生成的内容准确无误，避免任何可能出现的误导性陈述。\n",
       "- 面对道德和版权问题的复杂性，特别是在人工智能生成物的知识产权归属方面。\n",
       "\n",
       "通过分析上述案例，我们能够更好地理解 LLM 缩放定律在实际应用中的效果，同时也认识到在推进技术落地时所面临的问题与机遇。这些案例不仅展示了 LLM 的强大功能，还提醒我们在实践过程中需要综合考量多方面的因素。\n",
       "\n",
       "---\n",
       "\n",
       "# 未来发展与挑战\n",
       "\n",
       "## 概述\n",
       "\n",
       "为了更好地理解大型语言模型（LLM）的未来发展和面临的挑战，本章节将研究当前的缩放定律能够给我们提供哪些启示，并探讨未来的研究方向和面临的实际问题。当前，多项研究表明，模型规模的增加能够显著提升模型的性能，但这种增长并不是线性的。因此，我们期待LSTM未来仍能有重要的发展空间。为了推动LLM的发展，必须解决一系列技术问题和实际应用中的限制。\n",
       "\n",
       "## 技术限制\n",
       "\n",
       "首先，技术限制是推进LLM发展的主要障碍。训练大型神经网络需要大量的计算资源，这导致了高昂的硬件成本和能源消耗。此外，模型的训练和推理存在一定的复杂性，包括训练和推理的时间，以及如何有效管理硬件资源。\n",
       "\n",
       "## 实际应用中的挑战\n",
       "\n",
       "其次，实际应用中的挑战也是推动LLM未来发展的重要因素。一方面，如何提高模型的解释性和可理解性是目前大家关注的主要问题，实践中需要用户更好地理解和控制模型的运作过程。另一方面，LLM也需要具备应对各种边缘案例的能力，这涉及到数据分布、模型泛化能力和要求等诸多因素。\n",
       "\n",
       "## 未来方向\n",
       "\n",
       "当前的LLM研究存在许多值得关注的突破方向。第一，探索更高效而量化的训练方法，以降低硬件成本，并提高模型的可解释性。第二，增加模型的线上性（online ability）和泛化能力，能根据输入自动生成新的、高质量的文本。第三，采用适当的评估标准，用来评测LLM应用的工作质量。\n",
       "\n",
       "### 结论\n",
       "\n",
       "总之，考虑到未来的发展方向和存在的挑战，开发大型语言模型的科学家们应关注减少硬件成本和提高模型解释能力等重要因素。在未来，我们必须共同努力，以克服当前模型训练、推理和整合方面存在的技术限制，实现LLM更好的总体性能。\n",
       "\n",
       "---\n",
       "\n",
       "### 结论\n",
       "\n",
       "经过对 LLM（大型语言模型）进行深刻的探索和分析，本文得出了诸多重要的发现和见解：\n",
       "\n",
       "1. **模型规模与性能的强相关性**：通过实证研究，我们可以看到模型的参数数量与模型表现之间存在显著的正相关关系。这意味着更庞大的模型能够更好地捕捉语言的复杂性和细微差别，从而在各种自然语言处理任务上表现得更加出色。\n",
       "\n",
       "2. **算力利用率的提升**：尽管模型规模呈指数级增长，但实现有效利用算力提高训练效率的技术有了显著进步，使得在成本控制和时间效率之间达到一个较好的平衡。\n",
       "\n",
       "3. **计算资源的集中化影响**：超大规模语言模型的发展不可避免地加强了大型科技公司主导该领域的趋势，这些公司能够投入巨额资金来建立最先进的算力基础设施。\n",
       "\n",
       "4. **技术的双刃剑效应**：尽管 LLM 为许多行业带来了前所未有的机遇，但在技术进步的同时也引发了关于数据隐私、算法偏见和工作替换等一系列伦理和社会责任问题。对此，需要行业内外共同探讨深层次的改进策略。\n",
       "\n",
       "基于上述发现，LLM 的缩放定律对其行业具有重要的意义：\n",
       "\n",
       "1. **加速创新和技术进步的步伐**：随着模型规模的持续扩大，我们有望见证更多前沿技术的诞生，帮助企业增强竞争力以应对快速变化的市场环境。\n",
       "\n",
       "2. **推动合作与竞争的平衡**：虽然大公司可能因拥有更多的资源而形成优势，但小公司和初创企业同样能通过采用开源技术和合作平台找到自己的生存和发展空间。\n",
       "\n",
       "3. **促进公平发展**：对于缩小数字鸿沟而言，关键在于确保更多的开发者能够访问和使用这些先进技术而不仅仅是少数企业或个人。政府应当制定相应的政策鼓励创新，同时防止知识产权垄断情况的出现，确保资源能够惠及更广泛的社群。\n",
       "\n",
       "综上所述，LLM 的缩放定律不仅革新了自然语言处理领域，也为未来的技术发展和社会进步指明了方向。未来还将进一步探索 LLM 如何更好地满足人类的需求，促进多元化和社会和谐。"
      ],
      "text/plain": [
       "<IPython.core.display.Markdown object>"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from typing import List\n",
    "from pydantic import BaseModel, Field\n",
    "from langchain_openai import ChatOpenAI\n",
    "from langchain_core.messages import HumanMessage, SystemMessage\n",
    "from langgraph.func import entrypoint, task\n",
    "\n",
    "\n",
    "# 用于结构化输出的模式，用于规划报告章节\n",
    "class Section(BaseModel):\n",
    "    name: str = Field(\n",
    "        description=\"报告章节的名称\", # 报告章节的名称\n",
    "    )\n",
    "    description: str = Field(\n",
    "        description=\"本章节中要涵盖的主要主题和概念的简要概述\", # 章节内容的描述\n",
    "    )\n",
    "\n",
    "class Sections(BaseModel):\n",
    "    sections: List[Section] = Field(\n",
    "        description=\"报告的章节\", # 报告章节列表\n",
    "    )\n",
    "\n",
    "# 用于规划报告章节的增强型 LLM，使用结构化输出\n",
    "llm = ChatOpenAI(model=\"Qwen/Qwen2.5-7B-Instruct\")\n",
    "planner = llm.with_structured_output(Sections, method=\"function_calling\")\n",
    "\n",
    "@task\n",
    "def orchestrator(topic: str):\n",
    "    \"\"\"协调器，使用结构化输出生成报告计划\"\"\"\n",
    "    # 使用 planner LLM 和结构化输出生成报告章节计划\n",
    "    report_sections = planner.invoke(\n",
    "        [\n",
    "            SystemMessage(content=\"生成报告计划。\"), # planner LLM 的系统消息\n",
    "            HumanMessage(content=f\"这是报告主题：{topic}\"), # 包含报告主题的用户消息\n",
    "        ]\n",
    "    )\n",
    "    return report_sections.sections # 返回计划的章节\n",
    "\n",
    "@task\n",
    "def llm_call(section: Section):\n",
    "    \"\"\"工作者根据分配的章节详细信息编写报告章节\"\"\"\n",
    "    # 使用 LLM 根据章节名称和描述生成报告章节内容\n",
    "    result = llm.invoke(\n",
    "        [\n",
    "            SystemMessage(content=\"编写报告章节。\"), # 工作者 LLM 的系统消息\n",
    "            HumanMessage(\n",
    "                content=f\"这是章节名称：{section.name} 和描述：{section.description}\" # 包含章节详细信息的用户消息\n",
    "            ),\n",
    "        ]\n",
    "    )\n",
    "    return result.content # 返回生成的章节内容\n",
    "\n",
    "@task\n",
    "def synthesizer(completed_sections: list[str]):\n",
    "    \"\"\"从各个章节输出合成完整报告\"\"\"\n",
    "    # 将已完成章节格式化为单个字符串以用于最终报告\n",
    "    final_report = \"\\n\\n---\\n\\n\".join(completed_sections)\n",
    "    return final_report # 返回最终报告\n",
    "\n",
    "# 入口点装饰函数定义协调器-工作者工作流\n",
    "@entrypoint()\n",
    "def orchestrator_worker(topic: str):\n",
    "    sections = orchestrator(topic).result() # 执行协调器任务以获取报告章节计划\n",
    "    section_futures = [llm_call(section) for section in sections] # 并行动态创建和执行每个章节的工作者任务\n",
    "    final_report = synthesizer(\n",
    "        [section_fut.result() for section_fut in section_futures] # 在所有工作者任务完成后执行合成器任务\n",
    "    ).result() # 获取最终合成报告\n",
    "\n",
    "    return final_report # 返回最终报告\n",
    "\n",
    "# 使用示例报告主题调用协调器-工作者工作流\n",
    "report = orchestrator_worker.invoke(\"创建关于 LLM 缩放定律的报告\")\n",
    "\n",
    "from IPython.display import Markdown\n",
    "Markdown(report) # 以 Markdown 格式显示最终报告"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "20lpfn9iizf",
   "metadata": {},
   "source": [
    "### 7.1.6 评估器-优化器 (Evaluator-Optimizer)\n",
    "\n",
    "\"评估器-优化器\"工作流体现了一种迭代改进过程，模仿了人类通常通过反馈和修订来改进其工作的方式。在此模式中，一个增强型 LLM 调用负责生成初始响应，而另一个增强型 LLM 调用（\"评估器 (Evaluator)\"）的任务是提供对此响应的反馈。\n",
    "\n",
    "可以重复进行生成、评估和反馈的循环多次，直到获得令人满意的结果或达到预定义的迭代次数。当存在可以明确表达和评估的明确评估标准，并且迭代改进能够显著增加输出价值时，评估器-优化器工作流尤其有效。"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9djskjxb3nv",
   "metadata": {},
   "source": [
    "##### 示例 7-9：基于 Graph API 的\"评估器-优化器\"工作流实现\n",
    "\n",
    "让我们实现一个笑话生成和改进的评估器-优化器系统："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "bqncuzcyxkw",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "当然可以，这里有一个关于猫咪的笑话，希望你喜欢：\n",
      "\n",
      "为什么猫咪不喜欢玩闹娃娃？\n",
      "\n",
      "因为它们讨厌没有毛的东西！\n"
     ]
    }
   ],
   "source": [
    "from typing_extensions import TypedDict, Literal\n",
    "from pydantic import BaseModel, Field\n",
    "from langchain_openai import ChatOpenAI\n",
    "\n",
    "from langgraph.graph import StateGraph, START, END\n",
    "\n",
    "model = ChatOpenAI(model=\"Qwen/Qwen3-8B\")\n",
    "\n",
    "# 用于结构化输出的模式，用于评估，定义反馈结构\n",
    "class Feedback(BaseModel):\n",
    "    grade: Literal[\"funny\", \"not funny\"] = Field(\n",
    "        description=\"判断笑话是否有趣。\", # 评估等级：有趣或不好笑\n",
    "    )\n",
    "    feedback: str = Field(\n",
    "        description=\"如果笑话不好笑，请提供有关如何改进它的反馈。\", # 如果笑话不好笑，则提供关于如何改进它的反馈\n",
    "    )\n",
    "\n",
    "# 用于评估的增强型 LLM，配置为输出 Feedback 模式\n",
    "evaluator = llm.with_structured_output(Feedback, method=\"function_calling\")\n",
    "\n",
    "# 评估器-优化器工作流的图状态定义\n",
    "class State(TypedDict):\n",
    "    joke: str\n",
    "    topic: str\n",
    "    feedback: str\n",
    "    funny_or_not: str\n",
    "\n",
    "# 图中的节点\n",
    "def llm_call_generator(state: State):\n",
    "    \"\"\"LLM 生成笑话，可能会结合之前评估的反馈\"\"\"\n",
    "    if state.get(\"feedback\"): # 检查状态中是否存在反馈，指示之前的评估\n",
    "        msg = llm.invoke(\n",
    "            f\"写一个关于 {state['topic']} 的笑话，但要考虑反馈：{state['feedback']}\" # 调用 LLM 生成笑话，结合反馈\n",
    "        )\n",
    "    else:\n",
    "        msg = llm.invoke(f\"写一个关于 {state['topic']} 的笑话\") # 调用 LLM 生成初始笑话，不带反馈\n",
    "    return {\"joke\": msg.content} # 返回生成的笑话，更新状态中的 'joke'\n",
    "\n",
    "def llm_call_evaluator(state: State):\n",
    "    \"\"\"LLM 使用结构化输出评估生成的笑话\"\"\"\n",
    "    grade = evaluator.invoke(f\"评价笑话 {state['joke']}\") # 调用评估器 LLM 来评价笑话\n",
    "    return {\"funny_or_not\": grade.grade, \"feedback\": grade.feedback} # 返回评估等级和反馈，更新状态中的 'funny_or_not' 和 'feedback'\n",
    "\n",
    "# 条件边缘函数，用于根据评估结果进行路由，创建反馈循环\n",
    "def route_joke(state: State):\n",
    "    \"\"\"根据评估器的反馈，路由回笑话生成器或结束\"\"\"\n",
    "    if state[\"funny_or_not\"] == \"funny\": # 检查笑话是否被评估为有趣\n",
    "        return \"Accepted\" # 如果笑话被接受，则路由到结束\n",
    "    elif state[\"funny_or_not\"] == \"not funny\": # 检查笑话是否被评估为不好笑\n",
    "        return \"Rejected + Feedback\" # 如果笑话被拒绝，则路由回生成器以结合反馈\n",
    "\n",
    "# 构建评估器-优化器工作流，使用 StateGraph\n",
    "optimizer_builder = StateGraph(State)\n",
    "\n",
    "# 将节点添加到图中\n",
    "optimizer_builder.add_node(\"llm_call_generator\", llm_call_generator)\n",
    "optimizer_builder.add_node(\"llm_call_evaluator\", llm_call_evaluator)\n",
    "\n",
    "# 定义边缘以连接节点并建立反馈循环\n",
    "optimizer_builder.add_edge(START, \"llm_call_generator\") # 开始节点连接到笑话生成器\n",
    "optimizer_builder.add_edge(\"llm_call_generator\", \"llm_call_evaluator\") # 生成器节点连接到评估器节点\n",
    "optimizer_builder.add_conditional_edges(\n",
    "    \"llm_call_evaluator\",\n",
    "    route_joke,\n",
    "    {  # 由 route_joke 返回的名称：要访问的下一个节点的名称\n",
    "        \"Accepted\": END, # 如果笑话被接受，则路由到结束\n",
    "        \"Rejected + Feedback\": \"llm_call_generator\", # 如果笑话被拒绝，则路由回生成器，创建反馈循环\n",
    "    },\n",
    ")\n",
    "\n",
    "# 编译评估器-优化器工作流图\n",
    "optimizer_workflow = optimizer_builder.compile()\n",
    "\n",
    "# 使用示例主题调用评估器-优化器工作流\n",
    "state = optimizer_workflow.invoke({\"topic\": \"Cats\"})\n",
    "print(state[\"joke\"]) "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "98i0hb88t8p",
   "metadata": {},
   "source": [
    "**💡 评估器-优化器的关键机制**：\n",
    "\n",
    "- **迭代改进**：通过反馈循环不断优化输出质量\n",
    "- **结构化评估**：使用 Pydantic 模型确保评估的一致性\n",
    "- **条件路由**：根据评估结果决定是否继续优化\n",
    "- **状态记忆**：保持反馈信息以指导后续改进\n",
    "- **质量门控**：只有达到标准的输出才会被接受"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "xz8hbmaru4",
   "metadata": {},
   "source": [
    "##### 示例 7-10：基于 Functional API 的\"评估器-优化器\"工作流实现\n",
    "\n",
    "让我们用 Functional API 实现相同的评估器-优化器逻辑："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "0ta9y4a0uff",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "当然可以，这里有一个关于猫的轻松幽默小笑话：\n",
      "\n",
      "每当一只猫走进一家酒吧，\n",
      "它总是要求喝“鱼汤”而不是啤酒，\n",
      "吧台后面的人惊讶地问：“鱼汤是什么东西？”\n",
      "猫头转向窗外，指着对面的超市说：\n",
      "“看，就是那种装在大罐子里的淡黄色液体！”\n"
     ]
    }
   ],
   "source": [
    "from typing_extensions import Literal\n",
    "from pydantic import BaseModel, Field\n",
    "from langgraph.func import entrypoint, task\n",
    "\n",
    "\n",
    "# 用于结构化输出的模式，用于评估，定义反馈结构\n",
    "class Feedback(BaseModel):\n",
    "    grade: Literal[\"funny\", \"not funny\"] = Field(\n",
    "        description=\"判断笑话是否有趣。\", # 评估等级：有趣或不好笑\n",
    "    )\n",
    "    feedback: str = Field(\n",
    "        description=\"如果笑话不好笑，请提供有关如何改进它的反馈。\", # 如果笑话不好笑，则提供关于如何改进它的反馈\n",
    "    )\n",
    "\n",
    "# 用于评估的增强型 LLM，使用结构化输出\n",
    "evaluator = llm.with_structured_output(Feedback, method=\"function_calling\")\n",
    "\n",
    "# 工作流中的节点，定义为任务\n",
    "@task\n",
    "def llm_call_generator(topic: str, feedback: str = None):\n",
    "    \"\"\"LLM 生成笑话，可能会结合反馈\"\"\"\n",
    "    if feedback: # 检查是否提供了反馈\n",
    "        msg = llm.invoke(\n",
    "            f\"写一个关于 {topic} 的笑话，但要考虑反馈：{feedback}\" # 调用 LLM 生成笑话，结合反馈\n",
    "        )\n",
    "    else:\n",
    "        msg = llm.invoke(f\"写一个关于 {topic} 的笑话\") # 调用 LLM 生成初始笑话，不带反馈\n",
    "    return msg.content # 返回生成的笑话\n",
    "\n",
    "@task\n",
    "def llm_call_evaluator(joke: str):\n",
    "    \"\"\"LLM 使用结构化输出评估生成的笑话\"\"\"\n",
    "    feedback = evaluator.invoke(f\"评价笑话 {joke}\") # 调用评估器 LLM 来评价笑话\n",
    "    return feedback # 返回评估反馈\n",
    "\n",
    "# 入口点装饰函数定义评估器-优化器工作流\n",
    "@entrypoint()\n",
    "def optimizer_workflow(topic: str):\n",
    "    feedback = None # 将反馈初始化为 None，用于第一次迭代\n",
    "    max_iterations = 3 # 设置最大迭代次数以避免无限循环\n",
    "    iteration = 0\n",
    "    \n",
    "    while iteration < max_iterations: # 迭代反馈循环的开始\n",
    "        joke = llm_call_generator(topic, feedback).result() # 执行生成器任务以创建笑话\n",
    "        evaluation = llm_call_evaluator(joke).result() # 执行评估器任务以评价笑话并获取反馈\n",
    "        \n",
    "        if evaluation.grade == \"funny\": # 检查笑话是否被评估为有趣\n",
    "            return joke # 如果笑话有趣，则返回笑话\n",
    "        \n",
    "        feedback = evaluation.feedback # 更新反馈用于下一次迭代\n",
    "        iteration += 1\n",
    "        \n",
    "    return joke # 如果达到最大迭代次数，返回最后的笑话\n",
    "\n",
    "# 调用评估器-优化器工作流\n",
    "result = optimizer_workflow.invoke(\"Cats\")\n",
    "print(result)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "jlnhvab1dfl",
   "metadata": {},
   "source": [
    "## 7.2 LangGraph 中的多智能体架构\n",
    "\n",
    "随着我们在构建 AI 智能体方面不断进步，我们旨在解决的任务的复杂性通常需要超越单一的、单片式的智能体设计。多智能体系统的概念因此变得非常宝贵。我们不再依赖单个智能体来处理所有事情，而是将我们的 AI 应用程序分解为一组更小、更专注的智能体，每个智能体都具有特定的职责和专业知识。\n",
    "\n",
    "多智能体系统的核心优势在于：\n",
    "\n",
    "- **模块化**：将复杂的智能体分解为更小的、独立的智能体简化了开发、测试和维护\n",
    "- **专业化**：使我们能够创建针对特定领域或任务量身定制的专家智能体\n",
    "- **受控通信**：允许开发人员显式定义和管理智能体如何通信、交换信息以及协调其行动\n",
    "\n",
    "LangGraph 为构建和编排多智能体系统提供了强大的框架。在本节中，我们将探索几种关键多智能体架构。"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9mbea871jaq",
   "metadata": {},
   "source": [
    "### 7.2.1 主管架构\n",
    "\n",
    "主管（Supervisor）架构是多智能体系统中的核心模式，当需要明确的编排点来管理和指导任务流时，它尤其有效。在这种架构中，指定的主管智能体充当中央协调员的角色，监督和指导多个专业智能体的活动。\n",
    "\n",
    "LangChain 团队新推出的 LangGraph Supervisor 类库提供了一种简化的方式来创建基于主管的多智能体系统。"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4bs3i020xjc",
   "metadata": {},
   "source": [
    "##### 示例 7-11：使用 LangGraph Supervisor 类库搭建具有数学和研究智能体的主管架构\n",
    "\n",
    "首先安装必要的库，然后构建一个包含数学专家和研究专家的主管系统："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "wo6v33fpkkj",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "2024年FAANG公司（Meta、Apple、Amazon、Netflix、Alphabet）的总员工人数约为 **202.76万**。具体分布如下：\n",
      "\n",
      "- Meta: 67,317人  \n",
      "- Apple: 164,000人  \n",
      "- Amazon: 1,551,000人  \n",
      "- Netflix: 14,000人  \n",
      "- Alphabet: 181,269人  \n",
      "\n",
      "**总计**：202,760人（数据统计可能存在时间差或未公开最新信息，建议以公司官方发布为准）。\n"
     ]
    }
   ],
   "source": [
    "# 首先安装 langgraph-supervisor 库（如果尚未安装）\n",
    "# !pip install langgraph-supervisor\n",
    "\n",
    "from langchain_openai import ChatOpenAI\n",
    "from langgraph_supervisor import create_supervisor\n",
    "from langgraph.prebuilt import create_react_agent\n",
    "\n",
    "# 这里推荐使用 SiliconCloud 平台上参赛量较大且支持工具调用的付费模型\n",
    "model = ChatOpenAI(model=\"Qwen/Qwen3-8B\")\n",
    "\n",
    "# 1. 为专业智能体定义工具：\n",
    "# 定义表示每个专业智能体工具的函数。\n",
    "# 'add' 和 'multiply' 用于数学智能体，'web_search' 用于研究智能体。\n",
    "def add(a: float, b: float) -> float:\n",
    "    \"\"\"添加两个数字。\"\"\"\n",
    "    return a + b\n",
    "\n",
    "def multiply(a: float, b: float) -> float:\n",
    "    \"\"\"乘两个数字。\"\"\"\n",
    "    return a * b\n",
    "\n",
    "def web_search(query: str) -> str:\n",
    "    \"\"\"在网络上搜索信息。\"\"\"\n",
    "    return (\n",
    "        \"以下是 FAANG 公司 2024 年的员工人数：\\n\"\n",
    "        \"1. **Facebook (Meta)**: 67,317 名员工。\\n\"\n",
    "        \"2. **Apple**: 164,000 名员工。\\n\"\n",
    "        \"3. **Amazon**: 1,551,000 名员工。\\n\"\n",
    "        \"4. **Netflix**: 14,000 名员工。\\n\"\n",
    "        \"5. **Google (Alphabet)**: 181,269 名员工。\"\n",
    "    )\n",
    "\n",
    "# 2. 使用 create_react_agent 创建专业智能体：\n",
    "# 使用 LangGraph 的预构建 create_react_agent 创建每个专业智能体。\n",
    "# 每个智能体都配置了特定的模型、工具、名称和提示。\n",
    "math_agent = create_react_agent(\n",
    "    model=model,\n",
    "    tools=[add, multiply], # 数学智能体的工具是 'add' 和 'multiply'\n",
    "    name=\"math_expert\", # 标识数学智能体的名称\n",
    "    prompt=\"你是一名数学专家。始终一次使用一个工具。\" # 指导数学智能体行为的提示\n",
    ")\n",
    "\n",
    "research_agent = create_react_agent(\n",
    "    model=model,\n",
    "    tools=[web_search], # 研究智能体的工具是 'web_search'\n",
    "    name=\"research_expert\", # 标识研究智能体的名称\n",
    "    prompt=\"你是一名世界一流的研究员，可以访问网络搜索。不要做任何数学运算。\" # 指导研究智能体行为的提示\n",
    ")\n",
    "\n",
    "# 3. 使用 create_supervisor 创建主管工作流：\n",
    "# 使用 LangGraph Supervisor 的 create_supervisor 创建主管工作流。\n",
    "# 传递专业智能体列表、主管模型以及主管的提示。\n",
    "workflow = create_supervisor(\n",
    "    [research_agent, math_agent], # 由主管管理的专业智能体列表\n",
    "    model=model, # 主管智能体的模型\n",
    "    prompt=\"你是一名团队主管，管理着一名研究专家和一名数学专家。研究专家能够利用网络搜索工具进行查询。\", # 指导主管行为的提示\n",
    ")\n",
    "\n",
    "# 4. 编译并运行工作流：\n",
    "# 将工作流编译为 LangGraph 应用程序，并使用用户消息调用它。\n",
    "app = workflow.compile() # 将工作流编译为可执行的 LangGraph 应用程序\n",
    "result = app.invoke({ # 使用用户消息调用已编译的应用程序\n",
    "    \"messages\": [\n",
    "        {\n",
    "            \"role\": \"user\",\n",
    "            \"content\": \"2024 年 FAANG 公司的总员工人数是多少？\" # 用户关于 FAANG 公司员工人数的查询\n",
    "        }\n",
    "    ]\n",
    "})\n",
    "print(result[\"messages\"][-1].content)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6hq5q5mc0ad",
   "metadata": {},
   "source": [
    "### 7.2.2 层次化架构 (Hierarchical Architecture)\n",
    "\n",
    "层次化架构通过多层次的智能体组织来处理复杂任务。每层都有特定的职责，上层智能体负责高级决策，下层智能体执行具体任务。\n",
    "\n",
    "**核心特点：**\n",
    "- 多层次组织结构\n",
    "- 清晰的职责分工\n",
    "- 层次间的通信机制\n",
    "- 递归的任务分解"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "wcc56l188xg",
   "metadata": {},
   "source": [
    "##### 示例 7-12：具有研究和写作团队的分层主管系统"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "3gqhtd6h1es",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "2024年FAANG公司（Facebook/Instagram/Meta、Apple、Amazon、Netflix、Google/Alphabet）的总员工人数需根据各公司最新公开数据汇总，但需注意以下几点：\n",
      "\n",
      "1. **数据准确性**  \n",
      "   - Meta（原Facebook）：约67,000人（2024年Q2财报数据）  \n",
      "   - Apple：约164,000人（2024年财年员工人数）  \n",
      "   - Amazon：约1,551,000人（2024年Q1财报数据）  \n",
      "   - Netflix：约14,000人（2024年员工人数）  \n",
      "   - Google/Alphabet：约181,000人（2024年财季数据）  \n",
      "\n",
      "   **总计**：约 **2,577,000人**（可能存在统计口径差异，如总部与全球员工、季节性波动等）。\n",
      "\n",
      "2. **注意事项**  \n",
      "   - 各公司数据可能因统计时间（如季度 vs 年度）或合并范围不同而略有差异。  \n",
      "   - Amazon的员工人数显著高于其他公司，需确认是否包含其子公司（如AWS、Whole Foods等）。  \n",
      "   - Netflix的员工规模较小，主要因其订阅业务模式。  \n",
      "\n",
      "3. **建议**  \n",
      "   可通过各公司官网（如Meta、Apple、Amazon、Netflix、Google）或权威报告（如Statista、Glassdoor）核实最新数据，因实际数字可能随时间调整。  \n",
      "\n",
      "如需进一步分析（如员工增长趋势、部门分布等），可转交研究团队。\n"
     ]
    }
   ],
   "source": [
    "from langchain_openai import ChatOpenAI\n",
    "from langgraph_supervisor import create_supervisor\n",
    "from langgraph.prebuilt import create_react_agent\n",
    "\n",
    "# 这里推荐使用工具调用能力较强的大语言模型\n",
    "model = ChatOpenAI(model=\"Qwen/Qwen3-8B\")\n",
    "\n",
    "# --- 1. 定义研究团队智能体和主管 ---\n",
    "# 定义研究团队（math_expert 和 research_expert）的工具和智能体\n",
    "def add(a: float, b: float) -> float:\n",
    "    \"\"\"添加两个数字。\"\"\"\n",
    "    return a + b\n",
    "\n",
    "def multiply(a: float, b: float) -> float:\n",
    "    \"\"\"乘两个数字。\"\"\"\n",
    "    return a * b\n",
    "\n",
    "def web_search(query: str) -> str:\n",
    "    \"\"\"在网络上搜索信息。\"\"\"\n",
    "    return (\n",
    "        \"以下是 FAANG 公司 2024 年的员工人数：\\n\"\n",
    "        \"1. **Facebook (Meta)**: 67,317 名员工。\\n\"\n",
    "        \"2. **Apple**: 164,000 名员工。\\n\"\n",
    "        \"3. **Amazon**: 1,551,000 名员工。\\n\"\n",
    "        \"4. **Netflix**: 14,000 名员工。\\n\"\n",
    "        \"5. **Google (Alphabet)**: 181,269 名员工。\"\n",
    "    )\n",
    "\n",
    "math_agent = create_react_agent(\n",
    "    model=model,\n",
    "    tools=[add, multiply],\n",
    "    name=\"math_expert\",\n",
    "    prompt=\"你是一名数学专家。\"\n",
    ")\n",
    "\n",
    "research_agent = create_react_agent(\n",
    "    model=model,\n",
    "    tools=[web_search],\n",
    "    name=\"research_expert\",\n",
    "    prompt=\"你是一名世界一流的研究员，可以访问网络搜索。不要做任何数学运算。\"\n",
    ")\n",
    "\n",
    "research_team_supervisor = create_supervisor(\n",
    "    [research_agent, math_agent], # 研究团队内的智能体\n",
    "    model=model,\n",
    "    prompt=\"你正在管理一个研究团队，该团队由研究和数学专家组成。研究专家能够利用网络搜索工具进行查询。\", # 研究团队主管的提示\n",
    ")\n",
    "research_team = research_team_supervisor.compile(name=\"research_team\") # 编译研究团队工作流并命名\n",
    "\n",
    "# --- 2. 定义写作团队智能体和主管 ---\n",
    "# 定义写作团队（writing_expert 和 publishing_expert）的工具和智能体\n",
    "def write_report(topic: str) -> str:\n",
    "    \"\"\"撰写关于给定主题的报告。\"\"\"\n",
    "    return f\"关于 {topic} 的报告：... （报告的详细内容）\"\n",
    "\n",
    "def publish_report(report: str) -> str:\n",
    "    \"\"\"发布报告。\"\"\"\n",
    "    return f\"报告已发布：{report}\"\n",
    "\n",
    "writing_agent = create_react_agent(\n",
    "    model=model,\n",
    "    tools=[write_report],\n",
    "    name=\"writing_expert\",\n",
    "    prompt=\"你是一名写作专家。\"\n",
    ")\n",
    "\n",
    "publishing_agent = create_react_agent(\n",
    "    model=model,\n",
    "    tools=[publish_report],\n",
    "    name=\"publishing_expert\",\n",
    "    prompt=\"你是一名出版专家。\"\n",
    ")\n",
    "\n",
    "writing_team_supervisor = create_supervisor(\n",
    "    [writing_agent, publishing_agent], # 写作团队内的智能体\n",
    "    model=model,\n",
    "    prompt=\"你正在管理一个写作团队，该团队由写作和出版专家组成。\", # 写作团队主管的提示\n",
    ")\n",
    "writing_team = writing_team_supervisor.compile(name=\"writing_team\") # 编译写作团队工作流并命名\n",
    "\n",
    "# --- 3. 定义顶层主管 ---\n",
    "# 定义顶层主管以管理研究团队和写作团队\n",
    "top_level_supervisor_agent = create_supervisor(\n",
    "    [research_team, writing_team], # 传递已编译的研究和写作团队工作流作为智能体\n",
    "    model=model,\n",
    "    prompt=\"你是一名顶层主管，管理着研究团队和写作团队。\", # 顶层主管的提示\n",
    ")\n",
    "top_level_supervisor = top_level_supervisor_agent.compile(name=\"top_level_supervisor\") # 编译顶层主管工作流并命名\n",
    "\n",
    "# --- 4. 调用顶层主管 ---\n",
    "# 使用用户查询调用顶层主管\n",
    "result = top_level_supervisor.invoke({\n",
    "    \"messages\": [\n",
    "        {\n",
    "            \"role\": \"user\",\n",
    "            \"content\": \"2024 年 FAANG 公司的总员工人数是多少？\" # 用户查询\n",
    "        }\n",
    "    ]\n",
    "})\n",
    "print(result[\"messages\"][-1].content)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "hi459ml977",
   "metadata": {},
   "source": [
    "**💡 层次化架构关键特性**：\n",
    "\n",
    "- **清晰的层次结构**：高级管理层、中级管理层、执行层各司其职\n",
    "- **自上而下的任务分解**：从高级计划到详细任务再到具体执行\n",
    "- **动态工作者分配**：根据规划结果动态创建工作者节点\n",
    "- **层次间通信**：每层都有明确的输入输出接口\n",
    "- **递归管理结构**：可以进一步扩展为更深层次的管理结构"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "tqxdlfk4yxg",
   "metadata": {},
   "source": [
    "### 7.2.3 网络架构 (Network Architecture)\n",
    "\n",
    "网络架构（也称为群体架构 Swarm Architecture）允许智能体之间进行更灵活、自组织的交互。在这种架构中，智能体可以直接相互通信，形成动态的协作网络。\n",
    "\n",
    "**核心特点：**\n",
    "- 去中心化的协作模式\n",
    "- 智能体间的直接通信\n",
    "- 动态的角色分配\n",
    "- 自适应的工作流程"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "cqn061bww9a",
   "metadata": {},
   "source": [
    "##### 示例 7-13：Alice 和 Bob 的简单交互网络"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "mha9bt3xd6f",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "哼，哼！你找到我啦，臭小子！有什么事儿要和我这个老海盗说？是寻找宝藏，还是想听海上的奇闻异事？快说！\n",
      "\n",
      "\n",
      "5 + 7 等于12！\n"
     ]
    }
   ],
   "source": [
    "# 首先安装 langgraph-swarm 库\n",
    "# !pip install langgraph-swarm\n",
    "\n",
    "from langchain_openai import ChatOpenAI\n",
    "\n",
    "from langgraph.checkpoint.memory import InMemorySaver\n",
    "from langgraph.prebuilt import create_react_agent\n",
    "from langgraph_swarm import create_handoff_tool, create_swarm\n",
    "\n",
    "# 这里推荐使用 SiliconCloud 平台上参赛量较大且支持工具调用的付费模型\n",
    "model = ChatOpenAI(model=\"Qwen/Qwen3-8B\")\n",
    "\n",
    "def add(a: int, b: int) -> int:\n",
    "    \"\"\"添加两个数字\"\"\"\n",
    "    return a + b\n",
    "\n",
    "# 1. 创建专业智能体（Alice 和 Bob）：\n",
    "# 使用 create_react_agent 定义两个专业智能体 Alice 和 Bob。\n",
    "# Alice 是一位数学专家，拥有 'add' 工具和一个移交给 Bob 的移交工具。\n",
    "alice = create_react_agent(\n",
    "    model,\n",
    "    [add, create_handoff_tool(agent_name=\"Bob\")], # Alice 拥有 'add' 工具和移交给 Bob 的移交工具\n",
    "    prompt=\"你是 Alice，一位加法专家，使用工具完成所有加法。\",\n",
    "    name=\"Alice\",\n",
    ")\n",
    "\n",
    "# Bob 说话像个海盗，并拥有一个移交给 Alice 以寻求数学帮助的移交工具。\n",
    "bob = create_react_agent(\n",
    "    model,\n",
    "    [create_handoff_tool(agent_name=\"Alice\", description=\"务必将所有数学问题请转移给 Alice，她可以帮助解决数学问题\")], # Bob 拥有一个移交给 Alice 的移交工具\n",
    "    prompt=\"你是 Bob，你说话像个海盗。\",\n",
    "    name=\"Bob\",\n",
    ")\n",
    "\n",
    "# 2. 创建用于对话记忆的内存检查点：\n",
    "# InMemorySaver 用于短期记忆，这对于保持对话的连续性至关重要。\n",
    "checkpointer = InMemorySaver()\n",
    "\n",
    "# 3. 使用 create_swarm 创建 Swarm 工作流：\n",
    "# create_swarm 函数使用 Alice 和 Bob 设置 swarm 架构。\n",
    "# default_active_agent=\"Alice\" 将 Alice 设置为新对话的起始智能体。\n",
    "workflow = create_swarm(\n",
    "    [alice, bob], # swarm 中的智能体列表\n",
    "    default_active_agent=\"Alice\" # 用于启动新对话的默认智能体\n",
    ")\n",
    "\n",
    "# 4. 使用检查点编译工作流：\n",
    "# 编译 swarm 工作流，传递检查点以进行内存管理。\n",
    "app = workflow.compile(checkpointer=checkpointer)\n",
    "\n",
    "# 5. 在多轮对话中调用 Swarm：\n",
    "# 多次调用 swarm，模拟对话线程。\n",
    "# config={\"configurable\": {\"thread_id\": \"1\"}} 确保消息在同一对话线程中被跟踪。\n",
    "config = {\"configurable\": {\"thread_id\": \"1\"}}\n",
    "turn_1 = app.invoke( # 第一轮 - 用户想和 Bob 说话\n",
    "    {\"messages\": [{\"role\": \"user\", \"content\": \"我想和 Bob 说话\"}]},\n",
    "    config,\n",
    ")\n",
    "print(turn_1[\"messages\"][-1].content) # 第一轮的输出\n",
    "turn_2 = app.invoke( # 第二轮 - 用户提出一个数学问题\n",
    "    {\"messages\": [{\"role\": \"user\", \"content\": \"5 + 7 等于多少？\"}]},\n",
    "    config,\n",
    ")\n",
    "print(turn_2[\"messages\"][-1].content) # 第二轮的输出"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "kjpeq6s8dlo",
   "metadata": {},
   "source": [
    "## 7.3 情境感知智能体：后台主动式 AI 架构与模式\n",
    "\n",
    "传统的人工智能应用程序，尤其是那些利用大型语言模型 (LLM) 的应用程序，主要采用基于聊天的用户体验。虽然聊天模式的用户友好且易于实施，但它本质上将发起和管理交互的责任完全放在人类用户身上。\n",
    "\n",
    "一种变革性的替代方案在于情境感知智能体（Ambient Agent）架构。这些架构设想 AI 智能体在后台主动且持续地运行，勤勉地监控相关信息流并自主地对重要事件做出反应。"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "eu16upd6mrf",
   "metadata": {},
   "source": [
    "##### 示例 7-14：LangGraph 中可用的人机环路交互结构体\n",
    "\n",
    "让我们定义用于实现人机环路交互的数据结构："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "a70w65gnuvw",
   "metadata": {},
   "outputs": [],
   "source": [
    "from typing import TypedDict, Literal, Optional, Union\n",
    "\n",
    "# HumanInterruptConfig：定义人工中断操作的配置选项。\n",
    "class HumanInterruptConfig(TypedDict):\n",
    "    allow_ignore: bool  # allow_ignore：布尔值，是否允许用户忽略中断。\n",
    "    allow_respond: bool # allow_respond：布尔值，是否允许用户发送自由格式的响应。\n",
    "    allow_edit: bool    # allow_edit：布尔值，是否允许用户编辑操作参数。\n",
    "    allow_accept: bool  # allow_accept：布尔值，是否允许用户按原样接受操作。\n",
    "\n",
    "\n",
    "# ActionRequest：定义向人类请求的操作。\n",
    "class ActionRequest(TypedDict):\n",
    "    action: str      # action：字符串，操作的描述性名称或标题。\n",
    "    args: dict       # args：字典，与操作关联的参数（例如，工具调用参数）。\n",
    "\n",
    "\n",
    "# HumanInterrupt：表示人工中断请求的主要模式。\n",
    "class HumanInterrupt(TypedDict):\n",
    "    action_request: ActionRequest         # action_request：ActionRequest，有关请求操作的详细信息。\n",
    "    config: HumanInterruptConfig         # config：HumanInterruptConfig，人工响应的配置选项。\n",
    "    description: Optional[str]          # description：可选字符串，中断的详细描述，可以是 Markdown 格式。\n",
    "\n",
    "\n",
    "# HumanResponse：从 Agent Inbox UI 接收的人工响应模式。\n",
    "class HumanResponse(TypedDict):\n",
    "    type: Literal['accept', 'ignore', 'response', 'edit'] # type：Literal String，来自用户的响应类型（accept、ignore、response、edit）。\n",
    "    args: Union[None, str, ActionRequest]                # args：Union[None, String, ActionRequest]，与响应关联的参数，根据\"type\"而变化。\n",
    "                                                        #     - 对于\"accept\"和\"edit\"：ActionRequest，其中包含可能已修改的参数。\n",
    "                                                        #     - 对于\"response\"：包含用户文本响应的字符串。\n",
    "                                                        #     - 对于\"ignore\"：None。"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ie9v4jfm3",
   "metadata": {},
   "source": [
    "##### 示例 7-15：在 LangGraph 图函数中使用 HumanInterrupt 和 HumanResponse\n",
    "\n",
    "以下示例演示了如何在 LangGraph 工作流中实现人机环路交互："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "id": "hhw7p6t8hc",
   "metadata": {},
   "outputs": [],
   "source": [
    "from typing import TypedDict, Literal, Optional, Union\n",
    "from langgraph.graph import StateGraph, START, END\n",
    "from langgraph.types import interrupt\n",
    "from langchain_core.messages import ToolMessage, HumanMessage, AIMessage\n",
    "\n",
    "# 假设这些工具在其他地方定义\n",
    "tools_by_name = {\n",
    "    \"hypothetical_tool\": lambda **kwargs: f\"工具执行结果: {kwargs}\"\n",
    "}\n",
    "\n",
    "# 定义图状态，继承消息状态以用于对话历史记录\n",
    "class AgentState(TypedDict):\n",
    "    messages: list\n",
    "    tool_calls: list\n",
    "\n",
    "# 定义可能触发人工中断的图函数\n",
    "def agent_node(state: AgentState):\n",
    "    \"\"\"智能体节点，决定是否调用工具或请求人工输入。\"\"\"\n",
    "    messages = state[\"messages\"]\n",
    "    last_message = messages[-1] if messages else None\n",
    "\n",
    "    # 假设智能体决定调用工具并具有工具调用详细信息\n",
    "    tool_call_name = \"hypothetical_tool\"\n",
    "    tool_call_args = {\"input_arg\": \"example_value\"}\n",
    "\n",
    "    # 构建 HumanInterrupt 对象\n",
    "    request: HumanInterrupt = {\n",
    "        \"action_request\": {\n",
    "            \"action\": tool_call_name, # 操作名称是工具名称\n",
    "            \"args\": tool_call_args  # 操作参数是工具参数\n",
    "        },\n",
    "        \"config\": {\n",
    "            \"allow_ignore\": True, # 允许用户忽略工具调用\n",
    "            \"allow_respond\": True, # 允许用户提供自由格式的响应\n",
    "            \"allow_edit\": True,  # 允许用户编辑工具参数\n",
    "            \"allow_accept\": True # 允许用户按原样接受工具调用\n",
    "        },\n",
    "        \"description\": f\"智能体建议使用参数：`{tool_call_args}` 调用工具：`{tool_call_name}`。 你批准吗？\", # Agent Inbox UI 的描述\n",
    "    }\n",
    "\n",
    "    # 调用 interrupt 函数，在列表中传递 HumanInterrupt 请求\n",
    "    response_list = interrupt([request])\n",
    "    response = response_list[0] if response_list else None # 从列表中提取第一个响应\n",
    "\n",
    "    if response:\n",
    "        if response['type'] == \"accept\":\n",
    "            # 用户接受了工具调用，继续执行工具\n",
    "            tool_result = tools_by_name[tool_call_name](**response['args']['args']) # 使用（可能已修改的）参数执行工具\n",
    "            output_message = ToolMessage(content=str(tool_result), tool_call_id=\"example_id\") # 使用结果创建 ToolMessage\n",
    "        elif response['type'] == \"edit\":\n",
    "            # 用户编辑了工具调用参数，使用编辑后的参数执行工具\n",
    "            edited_args = response['args']['args'] # 从 ActionRequest 中提取编辑后的参数\n",
    "            tool_result = tools_by_name[tool_call_name](**edited_args) # 使用编辑后的参数执行工具\n",
    "            output_message = ToolMessage(content=str(tool_result), tool_call_id=\"example_id\") # 使用结果创建 ToolMessage\n",
    "        elif response['type'] == \"response\":\n",
    "            # 用户提供了文本响应，据此处理\n",
    "            user_response_text = response['args'] # 提取用户的文本响应\n",
    "            output_message = AIMessage(content=f\"用户响应：{user_response_text}。 根据响应继续进行。\") # 创建 AIMessage 以确认响应\n",
    "        elif response['type'] == \"ignore\":\n",
    "            # 用户忽略了中断，据此处理\n",
    "            output_message = AIMessage(content=\"人工中断被忽略。 继续进行，不进行工具调用。\") # 创建 AIMessage，指示中断被忽略\n",
    "        else:\n",
    "            output_message = AIMessage(content=\"未知的人工响应类型。\") # 处理意外的响应类型\n",
    "    else:\n",
    "        # 未收到响应（例如，中断处理超时或错误）\n",
    "        output_message = AIMessage(content=\"未收到人工响应，继续进行，不进行干预。\") # 处理未收到响应的情况\n",
    "\n",
    "    return {\"messages\": [output_message]} # 返回更新的消息状态\n",
    "\n",
    "\n",
    "# 构建 LangGraph 工作流\n",
    "builder = StateGraph(AgentState)\n",
    "builder.add_node(\"agent_step\", agent_node)\n",
    "builder.add_edge(START, \"agent_step\")\n",
    "builder.add_edge(\"agent_step\", END)\n",
    "\n",
    "# 编译工作流\n",
    "workflow = builder.compile()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "wb13gdsv69o",
   "metadata": {},
   "source": [
    "## 📚 本章总结\n",
    "\n",
    "通过本章的学习，我们深入探索了 AI 智能体系统的架构设计与范式应用，从基础的智能体工作流模式（提示链、路由、并行化、协调器-工作者、评估器-优化器）到复杂的多智能体系统架构（主管、层次化、网络架构），再到前瞻性的情境感知智能体架构。我们掌握了 LangGraph 中的关键技术实现，包括 Graph API 与 Functional API 的灵活应用、Send API 的动态工作者创建、结构化输出与状态管理、以及人机环路交互机制，为构建复杂、高效且可扩展的智能体应用系统奠定了坚实的技术基础。在下一章中，我们将进一步探讨智能体的高级应用场景和优化策略。"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": ".venv",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.13.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
